Add support for user partitioning based on cohort.
JIRA: TNL-710 IMPORTANT: this commit converts the course_groups package to using migrations. When deploying to an existing openedx instance, migration 0001 may fail with an error indicating that the CourseUserGroup table already exists. If this happens, running the 0001 migration first, with the --fake option, is recommended. After performing this step, remaining migrations should work as expected.
This commit is contained in:
jsa
0
openedx/core/djangoapps/course_groups/__init__.py
Normal file
0
openedx/core/djangoapps/course_groups/__init__.py
Normal file
389
openedx/core/djangoapps/course_groups/cohorts.py
Normal file
389
openedx/core/djangoapps/course_groups/cohorts.py
Normal file
@@ -0,0 +1,389 @@
|
||||
"""
|
||||
This file contains the logic for cohort groups, as exposed internally to the
|
||||
forums, and to the cohort admin views.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import random
|
||||
|
||||
from django.db.models.signals import post_save, m2m_changed
|
||||
from django.dispatch import receiver
|
||||
from django.http import Http404
|
||||
from django.utils.translation import ugettext as _
|
||||
|
||||
from courseware import courses
|
||||
from eventtracking import tracker
|
||||
from student.models import get_user_by_username_or_email
|
||||
from .models import CourseUserGroup, CourseUserGroupPartitionGroup
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@receiver(post_save, sender=CourseUserGroup)
|
||||
def _cohort_added(sender, **kwargs):
|
||||
"""Emits a tracking log event each time a cohort is created"""
|
||||
instance = kwargs["instance"]
|
||||
if kwargs["created"] and instance.group_type == CourseUserGroup.COHORT:
|
||||
tracker.emit(
|
||||
"edx.cohort.created",
|
||||
{"cohort_id": instance.id, "cohort_name": instance.name}
|
||||
)
|
||||
|
||||
|
||||
@receiver(m2m_changed, sender=CourseUserGroup.users.through)
|
||||
def _cohort_membership_changed(sender, **kwargs):
|
||||
"""Emits a tracking log event each time cohort membership is modified"""
|
||||
def get_event_iter(user_id_iter, cohort_iter):
|
||||
return (
|
||||
{"cohort_id": cohort.id, "cohort_name": cohort.name, "user_id": user_id}
|
||||
for user_id in user_id_iter
|
||||
for cohort in cohort_iter
|
||||
)
|
||||
|
||||
action = kwargs["action"]
|
||||
instance = kwargs["instance"]
|
||||
pk_set = kwargs["pk_set"]
|
||||
reverse = kwargs["reverse"]
|
||||
|
||||
if action == "post_add":
|
||||
event_name = "edx.cohort.user_added"
|
||||
elif action in ["post_remove", "pre_clear"]:
|
||||
event_name = "edx.cohort.user_removed"
|
||||
else:
|
||||
return
|
||||
|
||||
if reverse:
|
||||
user_id_iter = [instance.id]
|
||||
if action == "pre_clear":
|
||||
cohort_iter = instance.course_groups.filter(group_type=CourseUserGroup.COHORT)
|
||||
else:
|
||||
cohort_iter = CourseUserGroup.objects.filter(pk__in=pk_set, group_type=CourseUserGroup.COHORT)
|
||||
else:
|
||||
cohort_iter = [instance] if instance.group_type == CourseUserGroup.COHORT else []
|
||||
if action == "pre_clear":
|
||||
user_id_iter = (user.id for user in instance.users.all())
|
||||
else:
|
||||
user_id_iter = pk_set
|
||||
|
||||
for event in get_event_iter(user_id_iter, cohort_iter):
|
||||
tracker.emit(event_name, event)
|
||||
|
||||
|
||||
# A 'default cohort' is an auto-cohort that is automatically created for a course if no auto_cohort_groups have been
|
||||
# specified. It is intended to be used in a cohorted-course for users who have yet to be assigned to a cohort.
|
||||
# Note 1: If an administrator chooses to configure a cohort with the same name, the said cohort will be used as
|
||||
# the "default cohort".
|
||||
# Note 2: If auto_cohort_groups are configured after the 'default cohort' has been created and populated, the
|
||||
# stagnant 'default cohort' will still remain (now as a manual cohort) with its previously assigned students.
|
||||
# Translation Note: We are NOT translating this string since it is the constant identifier for the "default group"
|
||||
# and needed across product boundaries.
|
||||
DEFAULT_COHORT_NAME = "Default Group"
|
||||
|
||||
|
||||
class CohortAssignmentType(object):
|
||||
"""
|
||||
The various types of rule-based cohorts
|
||||
"""
|
||||
# No automatic rules are applied to this cohort; users must be manually added.
|
||||
NONE = "none"
|
||||
|
||||
# One of (possibly) multiple cohort groups to which users are randomly assigned.
|
||||
# Note: The 'default cohort' group is included in this category iff it exists and
|
||||
# there are no other random groups. (Also see Note 2 above.)
|
||||
RANDOM = "random"
|
||||
|
||||
@staticmethod
|
||||
def get(cohort, course):
|
||||
"""
|
||||
Returns the assignment type of the given cohort for the given course
|
||||
"""
|
||||
if cohort.name in course.auto_cohort_groups:
|
||||
return CohortAssignmentType.RANDOM
|
||||
elif len(course.auto_cohort_groups) == 0 and cohort.name == DEFAULT_COHORT_NAME:
|
||||
return CohortAssignmentType.RANDOM
|
||||
else:
|
||||
return CohortAssignmentType.NONE
|
||||
|
||||
|
||||
# tl;dr: global state is bad. capa reseeds random every time a problem is loaded. Even
|
||||
# if and when that's fixed, it's a good idea to have a local generator to avoid any other
|
||||
# code that messes with the global random module.
|
||||
_local_random = None
|
||||
|
||||
|
||||
def local_random():
|
||||
"""
|
||||
Get the local random number generator. In a function so that we don't run
|
||||
random.Random() at import time.
|
||||
"""
|
||||
# ironic, isn't it?
|
||||
global _local_random
|
||||
|
||||
if _local_random is None:
|
||||
_local_random = random.Random()
|
||||
|
||||
return _local_random
|
||||
|
||||
|
||||
def is_course_cohorted(course_key):
|
||||
"""
|
||||
Given a course key, return a boolean for whether or not the course is
|
||||
cohorted.
|
||||
|
||||
Raises:
|
||||
Http404 if the course doesn't exist.
|
||||
"""
|
||||
return courses.get_course_by_id(course_key).is_cohorted
|
||||
|
||||
|
||||
def get_cohort_id(user, course_key):
|
||||
"""
|
||||
Given a course key and a user, return the id of the cohort that user is
|
||||
assigned to in that course. If they don't have a cohort, return None.
|
||||
"""
|
||||
cohort = get_cohort(user, course_key)
|
||||
return None if cohort is None else cohort.id
|
||||
|
||||
|
||||
def is_commentable_cohorted(course_key, commentable_id):
|
||||
"""
|
||||
Args:
|
||||
course_key: CourseKey
|
||||
commentable_id: string
|
||||
|
||||
Returns:
|
||||
Bool: is this commentable cohorted?
|
||||
|
||||
Raises:
|
||||
Http404 if the course doesn't exist.
|
||||
"""
|
||||
course = courses.get_course_by_id(course_key)
|
||||
|
||||
if not course.is_cohorted:
|
||||
# this is the easy case :)
|
||||
ans = False
|
||||
elif commentable_id in course.top_level_discussion_topic_ids:
|
||||
# top level discussions have to be manually configured as cohorted
|
||||
# (default is not)
|
||||
ans = commentable_id in course.cohorted_discussions
|
||||
else:
|
||||
# inline discussions are cohorted by default
|
||||
ans = True
|
||||
|
||||
log.debug(u"is_commentable_cohorted({0}, {1}) = {2}".format(
|
||||
course_key, commentable_id, ans
|
||||
))
|
||||
return ans
|
||||
|
||||
|
||||
def get_cohorted_commentables(course_key):
|
||||
"""
|
||||
Given a course_key return a set of strings representing cohorted commentables.
|
||||
"""
|
||||
|
||||
course = courses.get_course_by_id(course_key)
|
||||
|
||||
if not course.is_cohorted:
|
||||
# this is the easy case :)
|
||||
ans = set()
|
||||
else:
|
||||
ans = course.cohorted_discussions
|
||||
|
||||
return ans
|
||||
|
||||
|
||||
def get_cohort(user, course_key):
|
||||
"""
|
||||
Given a Django user and a CourseKey, return the user's cohort in that
|
||||
cohort.
|
||||
|
||||
Arguments:
|
||||
user: a Django User object.
|
||||
course_key: CourseKey
|
||||
|
||||
Returns:
|
||||
A CourseUserGroup object if the course is cohorted and the User has a
|
||||
cohort, else None.
|
||||
|
||||
Raises:
|
||||
ValueError if the CourseKey doesn't exist.
|
||||
"""
|
||||
# First check whether the course is cohorted (users shouldn't be in a cohort
|
||||
# in non-cohorted courses, but settings can change after course starts)
|
||||
try:
|
||||
course = courses.get_course_by_id(course_key)
|
||||
except Http404:
|
||||
raise ValueError("Invalid course_key")
|
||||
|
||||
if not course.is_cohorted:
|
||||
return None
|
||||
|
||||
try:
|
||||
return CourseUserGroup.objects.get(
|
||||
course_id=course_key,
|
||||
group_type=CourseUserGroup.COHORT,
|
||||
users__id=user.id,
|
||||
)
|
||||
except CourseUserGroup.DoesNotExist:
|
||||
# Didn't find the group. We'll go on to create one if needed.
|
||||
pass
|
||||
|
||||
choices = course.auto_cohort_groups
|
||||
if len(choices) > 0:
|
||||
# Randomly choose one of the auto_cohort_groups, creating it if needed.
|
||||
group_name = local_random().choice(choices)
|
||||
else:
|
||||
# Use the "default cohort".
|
||||
group_name = DEFAULT_COHORT_NAME
|
||||
|
||||
group, __ = CourseUserGroup.objects.get_or_create(
|
||||
course_id=course_key,
|
||||
group_type=CourseUserGroup.COHORT,
|
||||
name=group_name
|
||||
)
|
||||
user.course_groups.add(group)
|
||||
return group
|
||||
|
||||
|
||||
def get_course_cohorts(course):
|
||||
"""
|
||||
Get a list of all the cohorts in the given course. This will include auto cohorts,
|
||||
regardless of whether or not the auto cohorts include any users.
|
||||
|
||||
Arguments:
|
||||
course: the course for which cohorts should be returned
|
||||
|
||||
Returns:
|
||||
A list of CourseUserGroup objects. Empty if there are no cohorts. Does
|
||||
not check whether the course is cohorted.
|
||||
"""
|
||||
# Ensure all auto cohorts are created.
|
||||
for group_name in course.auto_cohort_groups:
|
||||
CourseUserGroup.objects.get_or_create(
|
||||
course_id=course.location.course_key,
|
||||
group_type=CourseUserGroup.COHORT,
|
||||
name=group_name
|
||||
)
|
||||
|
||||
return list(CourseUserGroup.objects.filter(
|
||||
course_id=course.location.course_key,
|
||||
group_type=CourseUserGroup.COHORT
|
||||
))
|
||||
|
||||
### Helpers for cohort management views
|
||||
|
||||
|
||||
def get_cohort_by_name(course_key, name):
|
||||
"""
|
||||
Return the CourseUserGroup object for the given cohort. Raises DoesNotExist
|
||||
it isn't present.
|
||||
"""
|
||||
return CourseUserGroup.objects.get(
|
||||
course_id=course_key,
|
||||
group_type=CourseUserGroup.COHORT,
|
||||
name=name
|
||||
)
|
||||
|
||||
|
||||
def get_cohort_by_id(course_key, cohort_id):
|
||||
"""
|
||||
Return the CourseUserGroup object for the given cohort. Raises DoesNotExist
|
||||
it isn't present. Uses the course_key for extra validation...
|
||||
"""
|
||||
return CourseUserGroup.objects.get(
|
||||
course_id=course_key,
|
||||
group_type=CourseUserGroup.COHORT,
|
||||
id=cohort_id
|
||||
)
|
||||
|
||||
|
||||
def add_cohort(course_key, name):
|
||||
"""
|
||||
Add a cohort to a course. Raises ValueError if a cohort of the same name already
|
||||
exists.
|
||||
"""
|
||||
log.debug("Adding cohort %s to %s", name, course_key)
|
||||
if CourseUserGroup.objects.filter(course_id=course_key,
|
||||
group_type=CourseUserGroup.COHORT,
|
||||
name=name).exists():
|
||||
raise ValueError(_("You cannot create two cohorts with the same name"))
|
||||
|
||||
try:
|
||||
course = courses.get_course_by_id(course_key)
|
||||
except Http404:
|
||||
raise ValueError("Invalid course_key")
|
||||
|
||||
cohort = CourseUserGroup.objects.create(
|
||||
course_id=course.id,
|
||||
group_type=CourseUserGroup.COHORT,
|
||||
name=name
|
||||
)
|
||||
tracker.emit(
|
||||
"edx.cohort.creation_requested",
|
||||
{"cohort_name": cohort.name, "cohort_id": cohort.id}
|
||||
)
|
||||
return cohort
|
||||
|
||||
|
||||
def add_user_to_cohort(cohort, username_or_email):
|
||||
"""
|
||||
Look up the given user, and if successful, add them to the specified cohort.
|
||||
|
||||
Arguments:
|
||||
cohort: CourseUserGroup
|
||||
username_or_email: string. Treated as email if has '@'
|
||||
|
||||
Returns:
|
||||
Tuple of User object and string (or None) indicating previous cohort
|
||||
|
||||
Raises:
|
||||
User.DoesNotExist if can't find user.
|
||||
ValueError if user already present in this cohort.
|
||||
"""
|
||||
user = get_user_by_username_or_email(username_or_email)
|
||||
previous_cohort_name = None
|
||||
previous_cohort_id = None
|
||||
|
||||
course_cohorts = CourseUserGroup.objects.filter(
|
||||
course_id=cohort.course_id,
|
||||
users__id=user.id,
|
||||
group_type=CourseUserGroup.COHORT
|
||||
)
|
||||
if course_cohorts.exists():
|
||||
if course_cohorts[0] == cohort:
|
||||
raise ValueError("User {user_name} already present in cohort {cohort_name}".format(
|
||||
user_name=user.username,
|
||||
cohort_name=cohort.name
|
||||
))
|
||||
else:
|
||||
previous_cohort = course_cohorts[0]
|
||||
previous_cohort.users.remove(user)
|
||||
previous_cohort_name = previous_cohort.name
|
||||
previous_cohort_id = previous_cohort.id
|
||||
|
||||
tracker.emit(
|
||||
"edx.cohort.user_add_requested",
|
||||
{
|
||||
"user_id": user.id,
|
||||
"cohort_id": cohort.id,
|
||||
"cohort_name": cohort.name,
|
||||
"previous_cohort_id": previous_cohort_id,
|
||||
"previous_cohort_name": previous_cohort_name,
|
||||
}
|
||||
)
|
||||
cohort.users.add(user)
|
||||
return (user, previous_cohort_name)
|
||||
|
||||
|
||||
def get_partition_group_id_for_cohort(cohort):
|
||||
"""
|
||||
Get the ids of the partition and group to which this cohort has been linked
|
||||
as a tuple of (int, int).
|
||||
|
||||
If the cohort has not been linked to any partition/group, both values in the
|
||||
tuple will be None.
|
||||
"""
|
||||
res = CourseUserGroupPartitionGroup.objects.filter(course_user_group=cohort)
|
||||
if len(res):
|
||||
return res[0].partition_id, res[0].group_id
|
||||
return None, None
|
||||
@@ -0,0 +1,88 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import datetime
|
||||
from south.db import db
|
||||
from south.v2 import SchemaMigration
|
||||
from django.db import models
|
||||
|
||||
|
||||
class Migration(SchemaMigration):
|
||||
|
||||
def forwards(self, orm):
|
||||
# Adding model 'CourseUserGroup'
|
||||
db.create_table('course_groups_courseusergroup', (
|
||||
('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
|
||||
('name', self.gf('django.db.models.fields.CharField')(max_length=255)),
|
||||
('course_id', self.gf('xmodule_django.models.CourseKeyField')(max_length=255, db_index=True)),
|
||||
('group_type', self.gf('django.db.models.fields.CharField')(max_length=20)),
|
||||
))
|
||||
db.send_create_signal('course_groups', ['CourseUserGroup'])
|
||||
|
||||
# Adding unique constraint on 'CourseUserGroup', fields ['name', 'course_id']
|
||||
db.create_unique('course_groups_courseusergroup', ['name', 'course_id'])
|
||||
|
||||
# Adding M2M table for field users on 'CourseUserGroup'
|
||||
db.create_table('course_groups_courseusergroup_users', (
|
||||
('id', models.AutoField(verbose_name='ID', primary_key=True, auto_created=True)),
|
||||
('courseusergroup', models.ForeignKey(orm['course_groups.courseusergroup'], null=False)),
|
||||
('user', models.ForeignKey(orm['auth.user'], null=False))
|
||||
))
|
||||
db.create_unique('course_groups_courseusergroup_users', ['courseusergroup_id', 'user_id'])
|
||||
|
||||
def backwards(self, orm):
|
||||
# Removing unique constraint on 'CourseUserGroup', fields ['name', 'course_id']
|
||||
db.delete_unique('course_groups_courseusergroup', ['name', 'course_id'])
|
||||
|
||||
# Deleting model 'CourseUserGroup'
|
||||
db.delete_table('course_groups_courseusergroup')
|
||||
|
||||
# Removing M2M table for field users on 'CourseUserGroup'
|
||||
db.delete_table('course_groups_courseusergroup_users')
|
||||
|
||||
models = {
|
||||
'auth.group': {
|
||||
'Meta': {'object_name': 'Group'},
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '80'}),
|
||||
'permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'})
|
||||
},
|
||||
'auth.permission': {
|
||||
'Meta': {'ordering': "('content_type__app_label', 'content_type__model', 'codename')", 'unique_together': "(('content_type', 'codename'),)", 'object_name': 'Permission'},
|
||||
'codename': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
|
||||
'content_type': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['contenttypes.ContentType']"}),
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'name': ('django.db.models.fields.CharField', [], {'max_length': '50'})
|
||||
},
|
||||
'auth.user': {
|
||||
'Meta': {'object_name': 'User'},
|
||||
'date_joined': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
|
||||
'email': ('django.db.models.fields.EmailField', [], {'max_length': '75', 'blank': 'True'}),
|
||||
'first_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
|
||||
'groups': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Group']", 'symmetrical': 'False', 'blank': 'True'}),
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'is_active': ('django.db.models.fields.BooleanField', [], {'default': 'True'}),
|
||||
'is_staff': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
|
||||
'is_superuser': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
|
||||
'last_login': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
|
||||
'last_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
|
||||
'password': ('django.db.models.fields.CharField', [], {'max_length': '128'}),
|
||||
'user_permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'}),
|
||||
'username': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '30'})
|
||||
},
|
||||
'contenttypes.contenttype': {
|
||||
'Meta': {'ordering': "('name',)", 'unique_together': "(('app_label', 'model'),)", 'object_name': 'ContentType', 'db_table': "'django_content_type'"},
|
||||
'app_label': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'model': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
|
||||
'name': ('django.db.models.fields.CharField', [], {'max_length': '100'})
|
||||
},
|
||||
'course_groups.courseusergroup': {
|
||||
'Meta': {'unique_together': "(('name', 'course_id'),)", 'object_name': 'CourseUserGroup'},
|
||||
'course_id': ('xmodule_django.models.CourseKeyField', [], {'max_length': '255', 'db_index': 'True'}),
|
||||
'group_type': ('django.db.models.fields.CharField', [], {'max_length': '20'}),
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'name': ('django.db.models.fields.CharField', [], {'max_length': '255'}),
|
||||
'users': ('django.db.models.fields.related.ManyToManyField', [], {'db_index': 'True', 'related_name': "'course_groups'", 'symmetrical': 'False', 'to': "orm['auth.User']"})
|
||||
}
|
||||
}
|
||||
|
||||
complete_apps = ['course_groups']
|
||||
@@ -0,0 +1,82 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import datetime
|
||||
from south.db import db
|
||||
from south.v2 import SchemaMigration
|
||||
from django.db import models
|
||||
|
||||
|
||||
class Migration(SchemaMigration):
|
||||
|
||||
def forwards(self, orm):
|
||||
# Adding model 'CourseUserGroupPartitionGroup'
|
||||
db.create_table('course_groups_courseusergrouppartitiongroup', (
|
||||
('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
|
||||
('course_user_group', self.gf('django.db.models.fields.related.OneToOneField')(to=orm['course_groups.CourseUserGroup'], unique=True)),
|
||||
('partition_id', self.gf('django.db.models.fields.IntegerField')()),
|
||||
('group_id', self.gf('django.db.models.fields.IntegerField')()),
|
||||
('created_at', self.gf('django.db.models.fields.DateTimeField')(auto_now_add=True, blank=True)),
|
||||
('updated_at', self.gf('django.db.models.fields.DateTimeField')(auto_now=True, blank=True)),
|
||||
))
|
||||
db.send_create_signal('course_groups', ['CourseUserGroupPartitionGroup'])
|
||||
|
||||
def backwards(self, orm):
|
||||
# Deleting model 'CourseUserGroupPartitionGroup'
|
||||
db.delete_table('course_groups_courseusergrouppartitiongroup')
|
||||
|
||||
models = {
|
||||
'auth.group': {
|
||||
'Meta': {'object_name': 'Group'},
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '80'}),
|
||||
'permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'})
|
||||
},
|
||||
'auth.permission': {
|
||||
'Meta': {'ordering': "('content_type__app_label', 'content_type__model', 'codename')", 'unique_together': "(('content_type', 'codename'),)", 'object_name': 'Permission'},
|
||||
'codename': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
|
||||
'content_type': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['contenttypes.ContentType']"}),
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'name': ('django.db.models.fields.CharField', [], {'max_length': '50'})
|
||||
},
|
||||
'auth.user': {
|
||||
'Meta': {'object_name': 'User'},
|
||||
'date_joined': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
|
||||
'email': ('django.db.models.fields.EmailField', [], {'max_length': '75', 'blank': 'True'}),
|
||||
'first_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
|
||||
'groups': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Group']", 'symmetrical': 'False', 'blank': 'True'}),
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'is_active': ('django.db.models.fields.BooleanField', [], {'default': 'True'}),
|
||||
'is_staff': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
|
||||
'is_superuser': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
|
||||
'last_login': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
|
||||
'last_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
|
||||
'password': ('django.db.models.fields.CharField', [], {'max_length': '128'}),
|
||||
'user_permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'}),
|
||||
'username': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '30'})
|
||||
},
|
||||
'contenttypes.contenttype': {
|
||||
'Meta': {'ordering': "('name',)", 'unique_together': "(('app_label', 'model'),)", 'object_name': 'ContentType', 'db_table': "'django_content_type'"},
|
||||
'app_label': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'model': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
|
||||
'name': ('django.db.models.fields.CharField', [], {'max_length': '100'})
|
||||
},
|
||||
'course_groups.courseusergroup': {
|
||||
'Meta': {'unique_together': "(('name', 'course_id'),)", 'object_name': 'CourseUserGroup'},
|
||||
'course_id': ('xmodule_django.models.CourseKeyField', [], {'max_length': '255', 'db_index': 'True'}),
|
||||
'group_type': ('django.db.models.fields.CharField', [], {'max_length': '20'}),
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'name': ('django.db.models.fields.CharField', [], {'max_length': '255'}),
|
||||
'users': ('django.db.models.fields.related.ManyToManyField', [], {'db_index': 'True', 'related_name': "'course_groups'", 'symmetrical': 'False', 'to': "orm['auth.User']"})
|
||||
},
|
||||
'course_groups.courseusergrouppartitiongroup': {
|
||||
'Meta': {'object_name': 'CourseUserGroupPartitionGroup'},
|
||||
'course_user_group': ('django.db.models.fields.related.OneToOneField', [], {'to': "orm['course_groups.CourseUserGroup']", 'unique': 'True'}),
|
||||
'created_at': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'blank': 'True'}),
|
||||
'group_id': ('django.db.models.fields.IntegerField', [], {}),
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'partition_id': ('django.db.models.fields.IntegerField', [], {}),
|
||||
'updated_at': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'blank': 'True'})
|
||||
}
|
||||
}
|
||||
|
||||
complete_apps = ['course_groups']
|
||||
51
openedx/core/djangoapps/course_groups/models.py
Normal file
51
openedx/core/djangoapps/course_groups/models.py
Normal file
@@ -0,0 +1,51 @@
|
||||
import logging
|
||||
|
||||
from django.contrib.auth.models import User
|
||||
from django.db import models
|
||||
from xmodule_django.models import CourseKeyField
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CourseUserGroup(models.Model):
|
||||
"""
|
||||
This model represents groups of users in a course. Groups may have different types,
|
||||
which may be treated specially. For example, a user can be in at most one cohort per
|
||||
course, and cohorts are used to split up the forums by group.
|
||||
"""
|
||||
class Meta:
|
||||
unique_together = (('name', 'course_id'), )
|
||||
|
||||
name = models.CharField(max_length=255,
|
||||
help_text=("What is the name of this group? "
|
||||
"Must be unique within a course."))
|
||||
users = models.ManyToManyField(User, db_index=True, related_name='course_groups',
|
||||
help_text="Who is in this group?")
|
||||
|
||||
# Note: groups associated with particular runs of a course. E.g. Fall 2012 and Spring
|
||||
# 2013 versions of 6.00x will have separate groups.
|
||||
course_id = CourseKeyField(
|
||||
max_length=255,
|
||||
db_index=True,
|
||||
help_text="Which course is this group associated with?",
|
||||
)
|
||||
|
||||
# For now, only have group type 'cohort', but adding a type field to support
|
||||
# things like 'question_discussion', 'friends', 'off-line-class', etc
|
||||
COHORT = 'cohort'
|
||||
GROUP_TYPE_CHOICES = ((COHORT, 'Cohort'),)
|
||||
group_type = models.CharField(max_length=20, choices=GROUP_TYPE_CHOICES)
|
||||
|
||||
|
||||
class CourseUserGroupPartitionGroup(models.Model):
|
||||
"""
|
||||
"""
|
||||
course_user_group = models.OneToOneField(CourseUserGroup)
|
||||
partition_id = models.IntegerField(
|
||||
help_text="contains the id of a cohorted partition in this course"
|
||||
)
|
||||
group_id = models.IntegerField(
|
||||
help_text="contains the id of a specific group within the cohorted partition"
|
||||
)
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
updated_at = models.DateTimeField(auto_now=True)
|
||||
75
openedx/core/djangoapps/course_groups/partition_scheme.py
Normal file
75
openedx/core/djangoapps/course_groups/partition_scheme.py
Normal file
@@ -0,0 +1,75 @@
|
||||
"""
|
||||
Provides a UserPartition driver for cohorts.
|
||||
"""
|
||||
import logging
|
||||
|
||||
from .cohorts import get_cohort, get_partition_group_id_for_cohort
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CohortPartitionScheme(object):
|
||||
"""
|
||||
This scheme uses lms cohorts (CourseUserGroups) and cohort-partition
|
||||
mappings (CourseUserGroupPartitionGroup) to map lms users into Partition
|
||||
Groups.
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def get_group_for_user(cls, course_id, user, user_partition, track_function=None):
|
||||
"""
|
||||
Returns the Group from the specified user partition to which the user
|
||||
is assigned, via their cohort membership and any mappings from cohorts
|
||||
to partitions / groups that might exist.
|
||||
|
||||
If the user has not yet been assigned to a cohort, an assignment *might*
|
||||
be created on-the-fly, as determined by the course's cohort config.
|
||||
Any such side-effects will be triggered inside the call to
|
||||
cohorts.get_cohort().
|
||||
|
||||
If the user has no cohort mapping, or there is no (valid) cohort ->
|
||||
partition group mapping found, the function returns None.
|
||||
"""
|
||||
cohort = get_cohort(user, course_id)
|
||||
if cohort is None:
|
||||
# student doesn't have a cohort
|
||||
return None
|
||||
|
||||
partition_id, group_id = get_partition_group_id_for_cohort(cohort)
|
||||
if partition_id is None:
|
||||
# cohort isn't mapped to any partition group.
|
||||
return None
|
||||
|
||||
if partition_id != user_partition.id:
|
||||
# if we have a match but the partition doesn't match the requested
|
||||
# one it means the mapping is invalid. the previous state of the
|
||||
# partition configuration may have been modified.
|
||||
log.warn(
|
||||
"partition mismatch in CohortPartitionScheme: %r",
|
||||
{
|
||||
"requested_partition_id": user_partition.id,
|
||||
"found_partition_id": partition_id,
|
||||
"found_group_id": group_id,
|
||||
"cohort_id": cohort.id,
|
||||
}
|
||||
)
|
||||
# fail silently
|
||||
return None
|
||||
|
||||
group = user_partition.get_group(group_id)
|
||||
if group is None:
|
||||
# if we have a match but the group doesn't exist in the partition,
|
||||
# it means the mapping is invalid. the previous state of the
|
||||
# partition configuration may have been modified.
|
||||
log.warn(
|
||||
"group not found in CohortPartitionScheme: %r",
|
||||
{
|
||||
"requested_partition_id": user_partition.id,
|
||||
"requested_group_id": group_id,
|
||||
"cohort_id": cohort.id,
|
||||
}
|
||||
)
|
||||
# fail silently
|
||||
return None
|
||||
|
||||
return group
|
||||
91
openedx/core/djangoapps/course_groups/tests/helpers.py
Normal file
91
openedx/core/djangoapps/course_groups/tests/helpers.py
Normal file
@@ -0,0 +1,91 @@
|
||||
"""
|
||||
Helper methods for testing cohorts.
|
||||
"""
|
||||
from factory import post_generation, Sequence
|
||||
from factory.django import DjangoModelFactory
|
||||
from opaque_keys.edx.locations import SlashSeparatedCourseKey
|
||||
from xmodule.modulestore.django import modulestore
|
||||
from xmodule.modulestore import ModuleStoreEnum
|
||||
|
||||
from ..models import CourseUserGroup
|
||||
|
||||
|
||||
class CohortFactory(DjangoModelFactory):
|
||||
"""
|
||||
Factory for constructing mock cohorts.
|
||||
"""
|
||||
FACTORY_FOR = CourseUserGroup
|
||||
|
||||
name = Sequence("cohort{}".format)
|
||||
course_id = SlashSeparatedCourseKey("dummy", "dummy", "dummy")
|
||||
group_type = CourseUserGroup.COHORT
|
||||
|
||||
@post_generation
|
||||
def users(self, create, extracted, **kwargs): # pylint: disable=unused-argument
|
||||
"""
|
||||
Returns the users associated with the cohort.
|
||||
"""
|
||||
if extracted:
|
||||
self.users.add(*extracted)
|
||||
|
||||
|
||||
def topic_name_to_id(course, name):
|
||||
"""
|
||||
Given a discussion topic name, return an id for that name (includes
|
||||
course and url_name).
|
||||
"""
|
||||
return "{course}_{run}_{name}".format(
|
||||
course=course.location.course,
|
||||
run=course.url_name,
|
||||
name=name
|
||||
)
|
||||
|
||||
|
||||
def config_course_cohorts(
|
||||
course,
|
||||
discussions,
|
||||
cohorted,
|
||||
cohorted_discussions=None,
|
||||
auto_cohort_groups=None
|
||||
):
|
||||
"""
|
||||
Given a course with no discussion set up, add the discussions and set
|
||||
the cohort config appropriately.
|
||||
|
||||
Arguments:
|
||||
course: CourseDescriptor
|
||||
discussions: list of topic names strings. Picks ids and sort_keys
|
||||
automatically.
|
||||
cohorted: bool.
|
||||
cohorted_discussions: optional list of topic names. If specified,
|
||||
converts them to use the same ids as topic names.
|
||||
auto_cohort_groups: optional list of strings
|
||||
(names of groups to put students into).
|
||||
|
||||
Returns:
|
||||
Nothing -- modifies course in place.
|
||||
"""
|
||||
def to_id(name):
|
||||
return topic_name_to_id(course, name)
|
||||
|
||||
topics = dict((name, {"sort_key": "A",
|
||||
"id": to_id(name)})
|
||||
for name in discussions)
|
||||
|
||||
course.discussion_topics = topics
|
||||
|
||||
d = {"cohorted": cohorted}
|
||||
if cohorted_discussions is not None:
|
||||
d["cohorted_discussions"] = [to_id(name)
|
||||
for name in cohorted_discussions]
|
||||
|
||||
if auto_cohort_groups is not None:
|
||||
d["auto_cohort_groups"] = auto_cohort_groups
|
||||
|
||||
course.cohort_config = d
|
||||
|
||||
try:
|
||||
# Not implemented for XMLModulestore, which is used by test_cohorts.
|
||||
modulestore().update_item(course, ModuleStoreEnum.UserID.test)
|
||||
except NotImplementedError:
|
||||
pass
|
||||
658
openedx/core/djangoapps/course_groups/tests/test_cohorts.py
Normal file
658
openedx/core/djangoapps/course_groups/tests/test_cohorts.py
Normal file
@@ -0,0 +1,658 @@
|
||||
from django.conf import settings
|
||||
from django.contrib.auth.models import User
|
||||
from django.db import IntegrityError
|
||||
from django.http import Http404
|
||||
from django.test import TestCase
|
||||
from django.test.utils import override_settings
|
||||
from mock import call, patch
|
||||
|
||||
from opaque_keys.edx.locations import SlashSeparatedCourseKey
|
||||
from student.models import CourseEnrollment
|
||||
from student.tests.factories import UserFactory
|
||||
from xmodule.modulestore.django import modulestore, clear_existing_modulestores
|
||||
from xmodule.modulestore.tests.django_utils import TEST_DATA_MIXED_TOY_MODULESTORE, mixed_store_config
|
||||
|
||||
from ..models import CourseUserGroup, CourseUserGroupPartitionGroup
|
||||
from .. import cohorts
|
||||
from ..tests.helpers import topic_name_to_id, config_course_cohorts, CohortFactory
|
||||
|
||||
# NOTE: running this with the lms.envs.test config works without
|
||||
# manually overriding the modulestore. However, running with
|
||||
# cms.envs.test doesn't.
|
||||
|
||||
TEST_DATA_DIR = settings.COMMON_TEST_DATA_ROOT
|
||||
TEST_MAPPING = {'edX/toy/2012_Fall': 'xml'}
|
||||
TEST_DATA_MIXED_MODULESTORE = mixed_store_config(TEST_DATA_DIR, TEST_MAPPING)
|
||||
|
||||
|
||||
@patch("openedx.core.djangoapps.course_groups.cohorts.tracker")
|
||||
class TestCohortSignals(TestCase):
|
||||
def setUp(self):
|
||||
self.course_key = SlashSeparatedCourseKey("dummy", "dummy", "dummy")
|
||||
|
||||
def test_cohort_added(self, mock_tracker):
|
||||
# Add cohort
|
||||
cohort = CourseUserGroup.objects.create(
|
||||
name="TestCohort",
|
||||
course_id=self.course_key,
|
||||
group_type=CourseUserGroup.COHORT
|
||||
)
|
||||
mock_tracker.emit.assert_called_with(
|
||||
"edx.cohort.created",
|
||||
{"cohort_id": cohort.id, "cohort_name": cohort.name}
|
||||
)
|
||||
mock_tracker.reset_mock()
|
||||
|
||||
# Modify existing cohort
|
||||
cohort.name = "NewName"
|
||||
cohort.save()
|
||||
self.assertFalse(mock_tracker.called)
|
||||
|
||||
# Add non-cohort group
|
||||
CourseUserGroup.objects.create(
|
||||
name="TestOtherGroupType",
|
||||
course_id=self.course_key,
|
||||
group_type="dummy"
|
||||
)
|
||||
self.assertFalse(mock_tracker.called)
|
||||
|
||||
def test_cohort_membership_changed(self, mock_tracker):
|
||||
cohort_list = [CohortFactory() for _ in range(2)]
|
||||
non_cohort = CourseUserGroup.objects.create(
|
||||
name="dummy",
|
||||
course_id=self.course_key,
|
||||
group_type="dummy"
|
||||
)
|
||||
user_list = [UserFactory() for _ in range(2)]
|
||||
mock_tracker.reset_mock()
|
||||
|
||||
def assert_events(event_name_suffix, user_list, cohort_list):
|
||||
mock_tracker.emit.assert_has_calls([
|
||||
call(
|
||||
"edx.cohort.user_" + event_name_suffix,
|
||||
{
|
||||
"user_id": user.id,
|
||||
"cohort_id": cohort.id,
|
||||
"cohort_name": cohort.name,
|
||||
}
|
||||
)
|
||||
for user in user_list for cohort in cohort_list
|
||||
])
|
||||
|
||||
# Add users to cohort
|
||||
cohort_list[0].users.add(*user_list)
|
||||
assert_events("added", user_list, cohort_list[:1])
|
||||
mock_tracker.reset_mock()
|
||||
|
||||
# Remove users from cohort
|
||||
cohort_list[0].users.remove(*user_list)
|
||||
assert_events("removed", user_list, cohort_list[:1])
|
||||
mock_tracker.reset_mock()
|
||||
|
||||
# Clear users from cohort
|
||||
cohort_list[0].users.add(*user_list)
|
||||
cohort_list[0].users.clear()
|
||||
assert_events("removed", user_list, cohort_list[:1])
|
||||
mock_tracker.reset_mock()
|
||||
|
||||
# Clear users from non-cohort group
|
||||
non_cohort.users.add(*user_list)
|
||||
non_cohort.users.clear()
|
||||
self.assertFalse(mock_tracker.emit.called)
|
||||
|
||||
# Add cohorts to user
|
||||
user_list[0].course_groups.add(*cohort_list)
|
||||
assert_events("added", user_list[:1], cohort_list)
|
||||
mock_tracker.reset_mock()
|
||||
|
||||
# Remove cohorts from user
|
||||
user_list[0].course_groups.remove(*cohort_list)
|
||||
assert_events("removed", user_list[:1], cohort_list)
|
||||
mock_tracker.reset_mock()
|
||||
|
||||
# Clear cohorts from user
|
||||
user_list[0].course_groups.add(*cohort_list)
|
||||
user_list[0].course_groups.clear()
|
||||
assert_events("removed", user_list[:1], cohort_list)
|
||||
mock_tracker.reset_mock()
|
||||
|
||||
# Clear non-cohort groups from user
|
||||
user_list[0].course_groups.add(non_cohort)
|
||||
user_list[0].course_groups.clear()
|
||||
self.assertFalse(mock_tracker.emit.called)
|
||||
|
||||
|
||||
@override_settings(MODULESTORE=TEST_DATA_MIXED_TOY_MODULESTORE)
|
||||
class TestCohorts(TestCase):
|
||||
"""
|
||||
Test the cohorts feature
|
||||
"""
|
||||
def setUp(self):
|
||||
"""
|
||||
Make sure that course is reloaded every time--clear out the modulestore.
|
||||
"""
|
||||
clear_existing_modulestores()
|
||||
self.toy_course_key = SlashSeparatedCourseKey("edX", "toy", "2012_Fall")
|
||||
|
||||
def test_is_course_cohorted(self):
|
||||
"""
|
||||
Make sure cohorts.is_course_cohorted() correctly reports if a course is cohorted or not.
|
||||
"""
|
||||
course = modulestore().get_course(self.toy_course_key)
|
||||
self.assertFalse(course.is_cohorted)
|
||||
self.assertFalse(cohorts.is_course_cohorted(course.id))
|
||||
|
||||
config_course_cohorts(course, [], cohorted=True)
|
||||
|
||||
self.assertTrue(course.is_cohorted)
|
||||
self.assertTrue(cohorts.is_course_cohorted(course.id))
|
||||
|
||||
# Make sure we get a Http404 if there's no course
|
||||
fake_key = SlashSeparatedCourseKey('a', 'b', 'c')
|
||||
self.assertRaises(Http404, lambda: cohorts.is_course_cohorted(fake_key))
|
||||
|
||||
def test_get_cohort_id(self):
|
||||
"""
|
||||
Make sure that cohorts.get_cohort_id() correctly returns the cohort id, or raises a ValueError when given an
|
||||
invalid course key.
|
||||
"""
|
||||
course = modulestore().get_course(self.toy_course_key)
|
||||
self.assertFalse(course.is_cohorted)
|
||||
|
||||
user = UserFactory(username="test", email="a@b.com")
|
||||
self.assertIsNone(cohorts.get_cohort_id(user, course.id))
|
||||
|
||||
config_course_cohorts(course, discussions=[], cohorted=True)
|
||||
cohort = CohortFactory(course_id=course.id, name="TestCohort")
|
||||
cohort.users.add(user)
|
||||
self.assertEqual(cohorts.get_cohort_id(user, course.id), cohort.id)
|
||||
|
||||
self.assertRaises(
|
||||
ValueError,
|
||||
lambda: cohorts.get_cohort_id(user, SlashSeparatedCourseKey("course", "does_not", "exist"))
|
||||
)
|
||||
|
||||
def test_get_cohort(self):
|
||||
"""
|
||||
Make sure cohorts.get_cohort() does the right thing when the course is cohorted
|
||||
"""
|
||||
course = modulestore().get_course(self.toy_course_key)
|
||||
self.assertEqual(course.id, self.toy_course_key)
|
||||
self.assertFalse(course.is_cohorted)
|
||||
|
||||
user = UserFactory(username="test", email="a@b.com")
|
||||
other_user = UserFactory(username="test2", email="a2@b.com")
|
||||
|
||||
self.assertIsNone(cohorts.get_cohort(user, course.id), "No cohort created yet")
|
||||
|
||||
cohort = CohortFactory(course_id=course.id, name="TestCohort")
|
||||
cohort.users.add(user)
|
||||
|
||||
self.assertIsNone(
|
||||
cohorts.get_cohort(user, course.id),
|
||||
"Course isn't cohorted, so shouldn't have a cohort"
|
||||
)
|
||||
|
||||
# Make the course cohorted...
|
||||
config_course_cohorts(course, discussions=[], cohorted=True)
|
||||
|
||||
self.assertEquals(
|
||||
cohorts.get_cohort(user, course.id).id,
|
||||
cohort.id,
|
||||
"user should be assigned to the correct cohort"
|
||||
)
|
||||
self.assertEquals(
|
||||
cohorts.get_cohort(other_user, course.id).id,
|
||||
cohorts.get_cohort_by_name(course.id, cohorts.DEFAULT_COHORT_NAME).id,
|
||||
"other_user should be assigned to the default cohort"
|
||||
)
|
||||
|
||||
def test_auto_cohorting(self):
|
||||
"""
|
||||
Make sure cohorts.get_cohort() does the right thing with auto_cohort_groups
|
||||
"""
|
||||
course = modulestore().get_course(self.toy_course_key)
|
||||
self.assertFalse(course.is_cohorted)
|
||||
|
||||
user1 = UserFactory(username="test", email="a@b.com")
|
||||
user2 = UserFactory(username="test2", email="a2@b.com")
|
||||
user3 = UserFactory(username="test3", email="a3@b.com")
|
||||
user4 = UserFactory(username="test4", email="a4@b.com")
|
||||
|
||||
cohort = CohortFactory(course_id=course.id, name="TestCohort")
|
||||
|
||||
# user1 manually added to a cohort
|
||||
cohort.users.add(user1)
|
||||
|
||||
# Add an auto_cohort_group to the course...
|
||||
config_course_cohorts(
|
||||
course,
|
||||
discussions=[],
|
||||
cohorted=True,
|
||||
auto_cohort_groups=["AutoGroup"]
|
||||
)
|
||||
|
||||
self.assertEquals(cohorts.get_cohort(user1, course.id).id, cohort.id, "user1 should stay put")
|
||||
|
||||
self.assertEquals(cohorts.get_cohort(user2, course.id).name, "AutoGroup", "user2 should be auto-cohorted")
|
||||
|
||||
# Now make the auto_cohort_group list empty
|
||||
config_course_cohorts(
|
||||
course,
|
||||
discussions=[],
|
||||
cohorted=True,
|
||||
auto_cohort_groups=[]
|
||||
)
|
||||
|
||||
self.assertEquals(
|
||||
cohorts.get_cohort(user3, course.id).id,
|
||||
cohorts.get_cohort_by_name(course.id, cohorts.DEFAULT_COHORT_NAME).id,
|
||||
"No groups->default cohort"
|
||||
)
|
||||
|
||||
# Now set the auto_cohort_group to something different
|
||||
config_course_cohorts(
|
||||
course,
|
||||
discussions=[],
|
||||
cohorted=True,
|
||||
auto_cohort_groups=["OtherGroup"]
|
||||
)
|
||||
|
||||
self.assertEquals(
|
||||
cohorts.get_cohort(user4, course.id).name, "OtherGroup", "New list->new group"
|
||||
)
|
||||
self.assertEquals(
|
||||
cohorts.get_cohort(user1, course.id).name, "TestCohort", "user1 should still be in originally placed cohort"
|
||||
)
|
||||
self.assertEquals(
|
||||
cohorts.get_cohort(user2, course.id).name, "AutoGroup", "user2 should still be in originally placed cohort"
|
||||
)
|
||||
self.assertEquals(
|
||||
cohorts.get_cohort(user3, course.id).name,
|
||||
cohorts.get_cohort_by_name(course.id, cohorts.DEFAULT_COHORT_NAME).name,
|
||||
"user3 should still be in the default cohort"
|
||||
)
|
||||
|
||||
def test_auto_cohorting_randomization(self):
|
||||
"""
|
||||
Make sure cohorts.get_cohort() randomizes properly.
|
||||
"""
|
||||
course = modulestore().get_course(self.toy_course_key)
|
||||
self.assertFalse(course.is_cohorted)
|
||||
|
||||
groups = ["group_{0}".format(n) for n in range(5)]
|
||||
config_course_cohorts(
|
||||
course, discussions=[], cohorted=True, auto_cohort_groups=groups
|
||||
)
|
||||
|
||||
# Assign 100 users to cohorts
|
||||
for i in range(100):
|
||||
user = UserFactory(
|
||||
username="test_{0}".format(i),
|
||||
email="a@b{0}.com".format(i)
|
||||
)
|
||||
cohorts.get_cohort(user, course.id)
|
||||
|
||||
# Now make sure that the assignment was at least vaguely random:
|
||||
# each cohort should have at least 1, and fewer than 50 students.
|
||||
# (with 5 groups, probability of 0 users in any group is about
|
||||
# .8**100= 2.0e-10)
|
||||
for cohort_name in groups:
|
||||
cohort = cohorts.get_cohort_by_name(course.id, cohort_name)
|
||||
num_users = cohort.users.count()
|
||||
self.assertGreater(num_users, 1)
|
||||
self.assertLess(num_users, 50)
|
||||
|
||||
def test_get_course_cohorts_noop(self):
|
||||
"""
|
||||
Tests get_course_cohorts returns an empty list when no cohorts exist.
|
||||
"""
|
||||
course = modulestore().get_course(self.toy_course_key)
|
||||
config_course_cohorts(course, [], cohorted=True)
|
||||
self.assertEqual([], cohorts.get_course_cohorts(course))
|
||||
|
||||
def test_get_course_cohorts(self):
|
||||
"""
|
||||
Tests that get_course_cohorts returns all cohorts, including auto cohorts.
|
||||
"""
|
||||
course = modulestore().get_course(self.toy_course_key)
|
||||
config_course_cohorts(
|
||||
course, [], cohorted=True,
|
||||
auto_cohort_groups=["AutoGroup1", "AutoGroup2"]
|
||||
)
|
||||
|
||||
# add manual cohorts to course 1
|
||||
CohortFactory(course_id=course.id, name="ManualCohort")
|
||||
CohortFactory(course_id=course.id, name="ManualCohort2")
|
||||
|
||||
cohort_set = {c.name for c in cohorts.get_course_cohorts(course)}
|
||||
self.assertEqual(cohort_set, {"AutoGroup1", "AutoGroup2", "ManualCohort", "ManualCohort2"})
|
||||
|
||||
def test_is_commentable_cohorted(self):
|
||||
course = modulestore().get_course(self.toy_course_key)
|
||||
self.assertFalse(course.is_cohorted)
|
||||
|
||||
def to_id(name):
|
||||
return topic_name_to_id(course, name)
|
||||
|
||||
# no topics
|
||||
self.assertFalse(
|
||||
cohorts.is_commentable_cohorted(course.id, to_id("General")),
|
||||
"Course doesn't even have a 'General' topic"
|
||||
)
|
||||
|
||||
# not cohorted
|
||||
config_course_cohorts(course, ["General", "Feedback"], cohorted=False)
|
||||
|
||||
self.assertFalse(
|
||||
cohorts.is_commentable_cohorted(course.id, to_id("General")),
|
||||
"Course isn't cohorted"
|
||||
)
|
||||
|
||||
# cohorted, but top level topics aren't
|
||||
config_course_cohorts(course, ["General", "Feedback"], cohorted=True)
|
||||
|
||||
self.assertTrue(course.is_cohorted)
|
||||
self.assertFalse(
|
||||
cohorts.is_commentable_cohorted(course.id, to_id("General")),
|
||||
"Course is cohorted, but 'General' isn't."
|
||||
)
|
||||
self.assertTrue(
|
||||
cohorts.is_commentable_cohorted(course.id, to_id("random")),
|
||||
"Non-top-level discussion is always cohorted in cohorted courses."
|
||||
)
|
||||
|
||||
# cohorted, including "Feedback" top-level topics aren't
|
||||
config_course_cohorts(
|
||||
course, ["General", "Feedback"],
|
||||
cohorted=True,
|
||||
cohorted_discussions=["Feedback"]
|
||||
)
|
||||
|
||||
self.assertTrue(course.is_cohorted)
|
||||
self.assertFalse(
|
||||
cohorts.is_commentable_cohorted(course.id, to_id("General")),
|
||||
"Course is cohorted, but 'General' isn't."
|
||||
)
|
||||
self.assertTrue(
|
||||
cohorts.is_commentable_cohorted(course.id, to_id("Feedback")),
|
||||
"Feedback was listed as cohorted. Should be."
|
||||
)
|
||||
|
||||
def test_get_cohorted_commentables(self):
|
||||
"""
|
||||
Make sure cohorts.get_cohorted_commentables() correctly returns a list of strings representing cohorted
|
||||
commentables. Also verify that we can't get the cohorted commentables from a course which does not exist.
|
||||
"""
|
||||
course = modulestore().get_course(self.toy_course_key)
|
||||
|
||||
self.assertEqual(cohorts.get_cohorted_commentables(course.id), set())
|
||||
|
||||
config_course_cohorts(course, [], cohorted=True)
|
||||
self.assertEqual(cohorts.get_cohorted_commentables(course.id), set())
|
||||
|
||||
config_course_cohorts(
|
||||
course, ["General", "Feedback"],
|
||||
cohorted=True,
|
||||
cohorted_discussions=["Feedback"]
|
||||
)
|
||||
self.assertItemsEqual(
|
||||
cohorts.get_cohorted_commentables(course.id),
|
||||
set([topic_name_to_id(course, "Feedback")])
|
||||
)
|
||||
|
||||
config_course_cohorts(
|
||||
course, ["General", "Feedback"],
|
||||
cohorted=True,
|
||||
cohorted_discussions=["General", "Feedback"]
|
||||
)
|
||||
self.assertItemsEqual(
|
||||
cohorts.get_cohorted_commentables(course.id),
|
||||
set([topic_name_to_id(course, "General"), topic_name_to_id(course, "Feedback")])
|
||||
)
|
||||
self.assertRaises(
|
||||
Http404,
|
||||
lambda: cohorts.get_cohorted_commentables(SlashSeparatedCourseKey("course", "does_not", "exist"))
|
||||
)
|
||||
|
||||
def test_get_cohort_by_name(self):
|
||||
"""
|
||||
Make sure cohorts.get_cohort_by_name() properly finds a cohort by name for a given course. Also verify that it
|
||||
raises an error when the cohort is not found.
|
||||
"""
|
||||
course = modulestore().get_course(self.toy_course_key)
|
||||
|
||||
self.assertRaises(
|
||||
CourseUserGroup.DoesNotExist,
|
||||
lambda: cohorts.get_cohort_by_name(course.id, "CohortDoesNotExist")
|
||||
)
|
||||
|
||||
cohort = CohortFactory(course_id=course.id, name="MyCohort")
|
||||
|
||||
self.assertEqual(cohorts.get_cohort_by_name(course.id, "MyCohort"), cohort)
|
||||
|
||||
self.assertRaises(
|
||||
CourseUserGroup.DoesNotExist,
|
||||
lambda: cohorts.get_cohort_by_name(SlashSeparatedCourseKey("course", "does_not", "exist"), cohort)
|
||||
)
|
||||
|
||||
def test_get_cohort_by_id(self):
|
||||
"""
|
||||
Make sure cohorts.get_cohort_by_id() properly finds a cohort by id for a given
|
||||
course.
|
||||
"""
|
||||
course = modulestore().get_course(self.toy_course_key)
|
||||
cohort = CohortFactory(course_id=course.id, name="MyCohort")
|
||||
|
||||
self.assertEqual(cohorts.get_cohort_by_id(course.id, cohort.id), cohort)
|
||||
|
||||
cohort.delete()
|
||||
|
||||
self.assertRaises(
|
||||
CourseUserGroup.DoesNotExist,
|
||||
lambda: cohorts.get_cohort_by_id(course.id, cohort.id)
|
||||
)
|
||||
|
||||
@patch("openedx.core.djangoapps.course_groups.cohorts.tracker")
|
||||
def test_add_cohort(self, mock_tracker):
|
||||
"""
|
||||
Make sure cohorts.add_cohort() properly adds a cohort to a course and handles
|
||||
errors.
|
||||
"""
|
||||
course = modulestore().get_course(self.toy_course_key)
|
||||
added_cohort = cohorts.add_cohort(course.id, "My Cohort")
|
||||
mock_tracker.emit.assert_any_call(
|
||||
"edx.cohort.creation_requested",
|
||||
{"cohort_name": added_cohort.name, "cohort_id": added_cohort.id}
|
||||
)
|
||||
|
||||
self.assertEqual(added_cohort.name, "My Cohort")
|
||||
self.assertRaises(
|
||||
ValueError,
|
||||
lambda: cohorts.add_cohort(course.id, "My Cohort")
|
||||
)
|
||||
self.assertRaises(
|
||||
ValueError,
|
||||
lambda: cohorts.add_cohort(SlashSeparatedCourseKey("course", "does_not", "exist"), "My Cohort")
|
||||
)
|
||||
|
||||
@patch("openedx.core.djangoapps.course_groups.cohorts.tracker")
|
||||
def test_add_user_to_cohort(self, mock_tracker):
|
||||
"""
|
||||
Make sure cohorts.add_user_to_cohort() properly adds a user to a cohort and
|
||||
handles errors.
|
||||
"""
|
||||
course_user = UserFactory(username="Username", email="a@b.com")
|
||||
UserFactory(username="RandomUsername", email="b@b.com")
|
||||
course = modulestore().get_course(self.toy_course_key)
|
||||
CourseEnrollment.enroll(course_user, self.toy_course_key)
|
||||
first_cohort = CohortFactory(course_id=course.id, name="FirstCohort")
|
||||
second_cohort = CohortFactory(course_id=course.id, name="SecondCohort")
|
||||
|
||||
# Success cases
|
||||
# We shouldn't get back a previous cohort, since the user wasn't in one
|
||||
self.assertEqual(
|
||||
cohorts.add_user_to_cohort(first_cohort, "Username"),
|
||||
(course_user, None)
|
||||
)
|
||||
mock_tracker.emit.assert_any_call(
|
||||
"edx.cohort.user_add_requested",
|
||||
{
|
||||
"user_id": course_user.id,
|
||||
"cohort_id": first_cohort.id,
|
||||
"cohort_name": first_cohort.name,
|
||||
"previous_cohort_id": None,
|
||||
"previous_cohort_name": None,
|
||||
}
|
||||
)
|
||||
# Should get (user, previous_cohort_name) when moved from one cohort to
|
||||
# another
|
||||
self.assertEqual(
|
||||
cohorts.add_user_to_cohort(second_cohort, "Username"),
|
||||
(course_user, "FirstCohort")
|
||||
)
|
||||
mock_tracker.emit.assert_any_call(
|
||||
"edx.cohort.user_add_requested",
|
||||
{
|
||||
"user_id": course_user.id,
|
||||
"cohort_id": second_cohort.id,
|
||||
"cohort_name": second_cohort.name,
|
||||
"previous_cohort_id": first_cohort.id,
|
||||
"previous_cohort_name": first_cohort.name,
|
||||
}
|
||||
)
|
||||
# Error cases
|
||||
# Should get ValueError if user already in cohort
|
||||
self.assertRaises(
|
||||
ValueError,
|
||||
lambda: cohorts.add_user_to_cohort(second_cohort, "Username")
|
||||
)
|
||||
# UserDoesNotExist if user truly does not exist
|
||||
self.assertRaises(
|
||||
User.DoesNotExist,
|
||||
lambda: cohorts.add_user_to_cohort(first_cohort, "non_existent_username")
|
||||
)
|
||||
|
||||
|
||||
@override_settings(MODULESTORE=TEST_DATA_MIXED_MODULESTORE)
|
||||
class TestCohortsAndPartitionGroups(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
"""
|
||||
Regenerate a test course and cohorts for each test
|
||||
"""
|
||||
self.test_course_key = SlashSeparatedCourseKey("edX", "toy", "2012_Fall")
|
||||
self.course = modulestore().get_course(self.test_course_key)
|
||||
|
||||
self.first_cohort = CohortFactory(course_id=self.course.id, name="FirstCohort")
|
||||
self.second_cohort = CohortFactory(course_id=self.course.id, name="SecondCohort")
|
||||
|
||||
self.partition_id = 1
|
||||
self.group1_id = 10
|
||||
self.group2_id = 20
|
||||
|
||||
def _link_cohort_partition_group(self, cohort, partition_id, group_id):
|
||||
"""
|
||||
Utility to create cohort -> partition group assignments in the database.
|
||||
"""
|
||||
link = CourseUserGroupPartitionGroup(
|
||||
course_user_group=cohort,
|
||||
partition_id=partition_id,
|
||||
group_id=group_id,
|
||||
)
|
||||
link.save()
|
||||
return link
|
||||
|
||||
def test_get_partition_group_id_for_cohort(self):
|
||||
"""
|
||||
Basic test of the partition_group_id accessor function
|
||||
"""
|
||||
# api should return nothing for an unmapped cohort
|
||||
self.assertEqual(
|
||||
cohorts.get_partition_group_id_for_cohort(self.first_cohort),
|
||||
(None, None),
|
||||
)
|
||||
# create a link for the cohort in the db
|
||||
link = self._link_cohort_partition_group(
|
||||
self.first_cohort,
|
||||
self.partition_id,
|
||||
self.group1_id
|
||||
)
|
||||
# api should return the specified partition and group
|
||||
self.assertEqual(
|
||||
cohorts.get_partition_group_id_for_cohort(self.first_cohort),
|
||||
(self.partition_id, self.group1_id)
|
||||
)
|
||||
# delete the link in the db
|
||||
link.delete()
|
||||
# api should return nothing again
|
||||
self.assertEqual(
|
||||
cohorts.get_partition_group_id_for_cohort(self.first_cohort),
|
||||
(None, None),
|
||||
)
|
||||
|
||||
def test_multiple_cohorts(self):
|
||||
"""
|
||||
Test that multiple cohorts can be linked to the same partition group
|
||||
"""
|
||||
self._link_cohort_partition_group(
|
||||
self.first_cohort,
|
||||
self.partition_id,
|
||||
self.group1_id,
|
||||
)
|
||||
self._link_cohort_partition_group(
|
||||
self.second_cohort,
|
||||
self.partition_id,
|
||||
self.group1_id,
|
||||
)
|
||||
self.assertEqual(
|
||||
cohorts.get_partition_group_id_for_cohort(self.first_cohort),
|
||||
(self.partition_id, self.group1_id),
|
||||
)
|
||||
self.assertEqual(
|
||||
cohorts.get_partition_group_id_for_cohort(self.second_cohort),
|
||||
cohorts.get_partition_group_id_for_cohort(self.first_cohort),
|
||||
)
|
||||
|
||||
def test_multiple_partition_groups(self):
|
||||
"""
|
||||
Test that a cohort cannot be mapped to more than one partition group
|
||||
"""
|
||||
self._link_cohort_partition_group(
|
||||
self.first_cohort,
|
||||
self.partition_id,
|
||||
self.group1_id,
|
||||
)
|
||||
with self.assertRaisesRegexp(IntegrityError, 'not unique'):
|
||||
self._link_cohort_partition_group(
|
||||
self.first_cohort,
|
||||
self.partition_id,
|
||||
self.group2_id,
|
||||
)
|
||||
|
||||
def test_delete_cascade(self):
|
||||
"""
|
||||
Test that cohort -> partition group links are automatically deleted
|
||||
when their parent cohort is deleted.
|
||||
"""
|
||||
self._link_cohort_partition_group(
|
||||
self.first_cohort,
|
||||
self.partition_id,
|
||||
self.group1_id
|
||||
)
|
||||
self.assertEqual(
|
||||
cohorts.get_partition_group_id_for_cohort(self.first_cohort),
|
||||
(self.partition_id, self.group1_id)
|
||||
)
|
||||
# delete the link
|
||||
self.first_cohort.delete()
|
||||
# api should return nothing at that point
|
||||
self.assertEqual(
|
||||
cohorts.get_partition_group_id_for_cohort(self.first_cohort),
|
||||
(None, None),
|
||||
)
|
||||
# link should no longer exist because of delete cascade
|
||||
with self.assertRaises(CourseUserGroupPartitionGroup.DoesNotExist):
|
||||
CourseUserGroupPartitionGroup.objects.get(
|
||||
course_user_group_id=self.first_cohort.id
|
||||
)
|
||||
@@ -0,0 +1,257 @@
|
||||
"""
|
||||
Test the partitions and partitions service
|
||||
|
||||
"""
|
||||
|
||||
from django.conf import settings
|
||||
import django.test
|
||||
from django.test.utils import override_settings
|
||||
from mock import patch
|
||||
|
||||
from student.tests.factories import UserFactory
|
||||
from xmodule.partitions.partitions import Group, UserPartition, UserPartitionError
|
||||
from xmodule.modulestore.django import modulestore, clear_existing_modulestores
|
||||
from xmodule.modulestore.tests.django_utils import mixed_store_config
|
||||
from opaque_keys.edx.locations import SlashSeparatedCourseKey
|
||||
|
||||
from ..partition_scheme import CohortPartitionScheme
|
||||
from ..models import CourseUserGroupPartitionGroup
|
||||
from ..cohorts import add_user_to_cohort
|
||||
from .helpers import CohortFactory, config_course_cohorts
|
||||
|
||||
|
||||
TEST_DATA_DIR = settings.COMMON_TEST_DATA_ROOT
|
||||
TEST_MAPPING = {'edX/toy/2012_Fall': 'xml'}
|
||||
TEST_DATA_MIXED_MODULESTORE = mixed_store_config(TEST_DATA_DIR, TEST_MAPPING)
|
||||
|
||||
|
||||
@override_settings(MODULESTORE=TEST_DATA_MIXED_MODULESTORE)
|
||||
class TestCohortPartitionScheme(django.test.TestCase):
|
||||
"""
|
||||
Test the logic for linking a user to a partition group based on their cohort.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
"""
|
||||
Regenerate a course with cohort configuration, partition and groups,
|
||||
and a student for each test.
|
||||
"""
|
||||
self.course_key = SlashSeparatedCourseKey("edX", "toy", "2012_Fall")
|
||||
config_course_cohorts(modulestore().get_course(self.course_key), [], cohorted=True)
|
||||
|
||||
self.groups = [Group(10, 'Group 10'), Group(20, 'Group 20')]
|
||||
self.user_partition = UserPartition(
|
||||
0,
|
||||
'Test Partition',
|
||||
'for testing purposes',
|
||||
self.groups,
|
||||
scheme=CohortPartitionScheme
|
||||
)
|
||||
self.student = UserFactory.create()
|
||||
|
||||
def link_cohort_partition_group(self, cohort, partition, group):
|
||||
"""
|
||||
Utility for creating cohort -> partition group links
|
||||
"""
|
||||
CourseUserGroupPartitionGroup(
|
||||
course_user_group=cohort,
|
||||
partition_id=partition.id,
|
||||
group_id=group.id,
|
||||
).save()
|
||||
|
||||
def unlink_cohort_partition_group(self, cohort):
|
||||
"""
|
||||
Utility for removing cohort -> partition group links
|
||||
"""
|
||||
CourseUserGroupPartitionGroup.objects.filter(course_user_group=cohort).delete()
|
||||
|
||||
def assert_student_in_group(self, group, partition=None):
|
||||
"""
|
||||
Utility for checking that our test student comes up as assigned to the
|
||||
specified partition (or, if None, no partition at all)
|
||||
"""
|
||||
self.assertEqual(
|
||||
CohortPartitionScheme.get_group_for_user(
|
||||
self.course_key,
|
||||
self.student,
|
||||
partition or self.user_partition,
|
||||
),
|
||||
group
|
||||
)
|
||||
|
||||
def test_student_cohort_assignment(self):
|
||||
"""
|
||||
Test that the CohortPartitionScheme continues to return the correct
|
||||
group for a student as the student is moved in and out of different
|
||||
cohorts.
|
||||
"""
|
||||
first_cohort, second_cohort = [
|
||||
CohortFactory(course_id=self.course_key) for _ in range(2)
|
||||
]
|
||||
# place student 0 into first cohort
|
||||
add_user_to_cohort(first_cohort, self.student.username)
|
||||
self.assert_student_in_group(None)
|
||||
|
||||
# link first cohort to group 0 in the partition
|
||||
self.link_cohort_partition_group(
|
||||
first_cohort,
|
||||
self.user_partition,
|
||||
self.groups[0],
|
||||
)
|
||||
# link second cohort to to group 1 in the partition
|
||||
self.link_cohort_partition_group(
|
||||
second_cohort,
|
||||
self.user_partition,
|
||||
self.groups[1],
|
||||
)
|
||||
self.assert_student_in_group(self.groups[0])
|
||||
|
||||
# move student from first cohort to second cohort
|
||||
add_user_to_cohort(second_cohort, self.student.username)
|
||||
self.assert_student_in_group(self.groups[1])
|
||||
|
||||
# move the student out of the cohort
|
||||
second_cohort.users.remove(self.student)
|
||||
self.assert_student_in_group(None)
|
||||
|
||||
def test_cohort_partition_group_assignment(self):
|
||||
"""
|
||||
Test that the CohortPartitionScheme returns the correct group for a
|
||||
student in a cohort when the cohort link is created / moved / deleted.
|
||||
"""
|
||||
test_cohort = CohortFactory(course_id=self.course_key)
|
||||
|
||||
# assign user to cohort (but cohort isn't linked to a partition group yet)
|
||||
add_user_to_cohort(test_cohort, self.student.username)
|
||||
# scheme should not yet find any link
|
||||
self.assert_student_in_group(None)
|
||||
|
||||
# link cohort to group 0
|
||||
self.link_cohort_partition_group(
|
||||
test_cohort,
|
||||
self.user_partition,
|
||||
self.groups[0],
|
||||
)
|
||||
# now the scheme should find a link
|
||||
self.assert_student_in_group(self.groups[0])
|
||||
|
||||
# link cohort to group 1 (first unlink it from group 0)
|
||||
self.unlink_cohort_partition_group(test_cohort)
|
||||
self.link_cohort_partition_group(
|
||||
test_cohort,
|
||||
self.user_partition,
|
||||
self.groups[1],
|
||||
)
|
||||
# scheme should pick up the link
|
||||
self.assert_student_in_group(self.groups[1])
|
||||
|
||||
# unlink cohort from anywhere
|
||||
self.unlink_cohort_partition_group(
|
||||
test_cohort,
|
||||
)
|
||||
# scheme should now return nothing
|
||||
self.assert_student_in_group(None)
|
||||
|
||||
def setup_student_in_group_0(self):
|
||||
"""
|
||||
Utility to set up a cohort, add our student to the cohort, and link
|
||||
the cohort to self.groups[0]
|
||||
"""
|
||||
test_cohort = CohortFactory(course_id=self.course_key)
|
||||
|
||||
# link cohort to group 0
|
||||
self.link_cohort_partition_group(
|
||||
test_cohort,
|
||||
self.user_partition,
|
||||
self.groups[0],
|
||||
)
|
||||
# place student into cohort
|
||||
add_user_to_cohort(test_cohort, self.student.username)
|
||||
# check link is correct
|
||||
self.assert_student_in_group(self.groups[0])
|
||||
|
||||
def test_partition_changes_nondestructive(self):
|
||||
"""
|
||||
If the name of a user partition is changed, or a group is added to the
|
||||
partition, links from cohorts do not break.
|
||||
|
||||
If the name of a group is changed, links from cohorts do not break.
|
||||
"""
|
||||
self.setup_student_in_group_0()
|
||||
|
||||
# to simulate a non-destructive configuration change on the course, create
|
||||
# a new partition with the same id and scheme but with groups renamed and
|
||||
# a group added
|
||||
new_groups = [Group(10, 'New Group 10'), Group(20, 'New Group 20'), Group(30, 'New Group 30')]
|
||||
new_user_partition = UserPartition(
|
||||
0, # same id
|
||||
'Different Partition',
|
||||
'dummy',
|
||||
new_groups,
|
||||
scheme=CohortPartitionScheme,
|
||||
)
|
||||
# the link should still work
|
||||
self.assert_student_in_group(new_groups[0], new_user_partition)
|
||||
|
||||
def test_missing_group(self):
|
||||
"""
|
||||
If the group is deleted (or its id is changed), there's no referential
|
||||
integrity enforced, so any references from cohorts to that group will be
|
||||
lost. A warning should be logged when links are found from cohorts to
|
||||
groups that no longer exist.
|
||||
"""
|
||||
self.setup_student_in_group_0()
|
||||
|
||||
# to simulate a destructive change on the course, create a new partition
|
||||
# with the same id, but different group ids.
|
||||
new_user_partition = UserPartition(
|
||||
0, # same id
|
||||
'Another Partition',
|
||||
'dummy',
|
||||
[Group(11, 'Not Group 10'), Group(21, 'Not Group 20')], # different ids
|
||||
scheme=CohortPartitionScheme,
|
||||
)
|
||||
# the partition will be found since it has the same id, but the group
|
||||
# ids aren't present anymore, so the scheme returns None (and logs a
|
||||
# warning)
|
||||
with patch('openedx.core.djangoapps.course_groups.partition_scheme.log') as mock_log:
|
||||
self.assert_student_in_group(None, new_user_partition)
|
||||
self.assertTrue(mock_log.warn.called)
|
||||
self.assertRegexpMatches(mock_log.warn.call_args[0][0], 'group not found')
|
||||
|
||||
def test_missing_partition(self):
|
||||
"""
|
||||
If the user partition is deleted (or its id is changed), there's no
|
||||
referential integrity enforced, so any references from cohorts to that
|
||||
partition's groups will be lost. A warning should be logged when links
|
||||
are found from cohorts to partitions that do not exist.
|
||||
"""
|
||||
self.setup_student_in_group_0()
|
||||
|
||||
# to simulate another destructive change on the course, create a new
|
||||
# partition with a different id, but using the same groups.
|
||||
new_user_partition = UserPartition(
|
||||
1, # different id
|
||||
'Moved Partition',
|
||||
'dummy',
|
||||
[Group(10, 'Group 10'), Group(20, 'Group 20')], # same ids
|
||||
scheme=CohortPartitionScheme,
|
||||
)
|
||||
# the partition will not be found even though the group ids match, so the
|
||||
# scheme returns None (and logs a warning).
|
||||
with patch('openedx.core.djangoapps.course_groups.partition_scheme.log') as mock_log:
|
||||
self.assert_student_in_group(None, new_user_partition)
|
||||
self.assertTrue(mock_log.warn.called)
|
||||
self.assertRegexpMatches(mock_log.warn.call_args[0][0], 'partition mismatch')
|
||||
|
||||
|
||||
class TestExtension(django.test.TestCase):
|
||||
"""
|
||||
Ensure that the scheme extension is correctly plugged in (via entry point
|
||||
in setup.py)
|
||||
"""
|
||||
|
||||
def test_get_scheme(self):
|
||||
self.assertEqual(UserPartition.get_scheme('cohort'), CohortPartitionScheme)
|
||||
with self.assertRaisesRegexp(UserPartitionError, 'Unrecognized scheme'):
|
||||
UserPartition.get_scheme('other')
|
||||
809
openedx/core/djangoapps/course_groups/tests/test_views.py
Normal file
809
openedx/core/djangoapps/course_groups/tests/test_views.py
Normal file
@@ -0,0 +1,809 @@
|
||||
"""
|
||||
Tests for course group views
|
||||
"""
|
||||
from collections import namedtuple
|
||||
import json
|
||||
|
||||
from collections import namedtuple
|
||||
from django.contrib.auth.models import User
|
||||
from django.http import Http404
|
||||
from django.test.client import RequestFactory
|
||||
from django.test.utils import override_settings
|
||||
|
||||
from xmodule.modulestore.tests.django_utils import TEST_DATA_MOCK_MODULESTORE
|
||||
from student.models import CourseEnrollment
|
||||
from student.tests.factories import UserFactory
|
||||
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
|
||||
from xmodule.modulestore.tests.factories import CourseFactory
|
||||
from opaque_keys.edx.locations import SlashSeparatedCourseKey
|
||||
|
||||
from ..models import CourseUserGroup
|
||||
from ..views import list_cohorts, add_cohort, users_in_cohort, add_users_to_cohort, remove_user_from_cohort
|
||||
from ..cohorts import get_cohort, CohortAssignmentType, get_cohort_by_name, DEFAULT_COHORT_NAME
|
||||
from .helpers import config_course_cohorts, CohortFactory
|
||||
|
||||
|
||||
@override_settings(MODULESTORE=TEST_DATA_MOCK_MODULESTORE)
|
||||
class CohortViewsTestCase(ModuleStoreTestCase):
|
||||
"""
|
||||
Base class which sets up a course and staff/non-staff users.
|
||||
"""
|
||||
def setUp(self):
|
||||
self.course = CourseFactory.create()
|
||||
self.staff_user = UserFactory(is_staff=True, username="staff")
|
||||
self.non_staff_user = UserFactory(username="nonstaff")
|
||||
|
||||
def _enroll_users(self, users, course_key):
|
||||
"""Enroll each user in the specified course"""
|
||||
for user in users:
|
||||
CourseEnrollment.enroll(user, course_key)
|
||||
|
||||
def _create_cohorts(self):
|
||||
"""Creates cohorts for testing"""
|
||||
self.cohort1_users = [UserFactory() for _ in range(3)]
|
||||
self.cohort2_users = [UserFactory() for _ in range(2)]
|
||||
self.cohort3_users = [UserFactory() for _ in range(2)]
|
||||
self.cohortless_users = [UserFactory() for _ in range(3)]
|
||||
self.unenrolled_users = [UserFactory() for _ in range(3)]
|
||||
self._enroll_users(
|
||||
self.cohort1_users + self.cohort2_users + self.cohort3_users + self.cohortless_users,
|
||||
self.course.id
|
||||
)
|
||||
self.cohort1 = CohortFactory(course_id=self.course.id, users=self.cohort1_users)
|
||||
self.cohort2 = CohortFactory(course_id=self.course.id, users=self.cohort2_users)
|
||||
self.cohort3 = CohortFactory(course_id=self.course.id, users=self.cohort3_users)
|
||||
|
||||
def _user_in_cohort(self, username, cohort):
|
||||
"""
|
||||
Return true iff a user with `username` exists in `cohort`.
|
||||
"""
|
||||
return username in [user.username for user in cohort.users.all()]
|
||||
|
||||
def _verify_non_staff_cannot_access(self, view, request_method, view_args):
|
||||
"""
|
||||
Verify that a non-staff user cannot access a given view.
|
||||
|
||||
`view` is the view to test.
|
||||
`view_args` is a list of arguments (not including the request) to pass
|
||||
to the view.
|
||||
"""
|
||||
if request_method == "GET":
|
||||
request = RequestFactory().get("dummy_url")
|
||||
elif request_method == "POST":
|
||||
request = RequestFactory().post("dummy_url")
|
||||
else:
|
||||
request = RequestFactory().request()
|
||||
request.user = self.non_staff_user
|
||||
view_args.insert(0, request)
|
||||
self.assertRaises(Http404, view, *view_args)
|
||||
|
||||
|
||||
class ListCohortsTestCase(CohortViewsTestCase):
|
||||
"""
|
||||
Tests the `list_cohorts` view.
|
||||
"""
|
||||
def request_list_cohorts(self, course):
|
||||
"""
|
||||
Call `list_cohorts` for a given `course` and return its response as a
|
||||
dict.
|
||||
"""
|
||||
request = RequestFactory().get("dummy_url")
|
||||
request.user = self.staff_user
|
||||
response = list_cohorts(request, course.id.to_deprecated_string())
|
||||
self.assertEqual(response.status_code, 200)
|
||||
return json.loads(response.content)
|
||||
|
||||
def verify_lists_expected_cohorts(self, expected_cohorts, response_dict=None):
|
||||
"""
|
||||
Verify that the server response contains the expected_cohorts.
|
||||
If response_dict is None, the list of cohorts is requested from the server.
|
||||
"""
|
||||
if response_dict is None:
|
||||
response_dict = self.request_list_cohorts(self.course)
|
||||
|
||||
self.assertTrue(response_dict.get("success"))
|
||||
self.assertItemsEqual(
|
||||
response_dict.get("cohorts"),
|
||||
[
|
||||
{
|
||||
"name": cohort.name,
|
||||
"id": cohort.id,
|
||||
"user_count": cohort.user_count,
|
||||
"assignment_type": cohort.assignment_type
|
||||
}
|
||||
for cohort in expected_cohorts
|
||||
]
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def create_expected_cohort(cohort, user_count, assignment_type):
|
||||
"""
|
||||
Create a tuple storing the expected cohort information.
|
||||
"""
|
||||
cohort_tuple = namedtuple("Cohort", "name id user_count assignment_type")
|
||||
return cohort_tuple(
|
||||
name=cohort.name, id=cohort.id, user_count=user_count, assignment_type=assignment_type
|
||||
)
|
||||
|
||||
def test_non_staff(self):
|
||||
"""
|
||||
Verify that we cannot access list_cohorts if we're a non-staff user.
|
||||
"""
|
||||
self._verify_non_staff_cannot_access(list_cohorts, "GET", [self.course.id.to_deprecated_string()])
|
||||
|
||||
def test_no_cohorts(self):
|
||||
"""
|
||||
Verify that no cohorts are in response for a course with no cohorts.
|
||||
"""
|
||||
self.verify_lists_expected_cohorts([])
|
||||
|
||||
def test_some_cohorts(self):
|
||||
"""
|
||||
Verify that cohorts are in response for a course with some cohorts.
|
||||
"""
|
||||
self._create_cohorts()
|
||||
expected_cohorts = [
|
||||
ListCohortsTestCase.create_expected_cohort(self.cohort1, 3, CohortAssignmentType.NONE),
|
||||
ListCohortsTestCase.create_expected_cohort(self.cohort2, 2, CohortAssignmentType.NONE),
|
||||
ListCohortsTestCase.create_expected_cohort(self.cohort3, 2, CohortAssignmentType.NONE),
|
||||
]
|
||||
self.verify_lists_expected_cohorts(expected_cohorts)
|
||||
|
||||
def test_auto_cohorts(self):
|
||||
"""
|
||||
Verify that auto cohorts are included in the response.
|
||||
"""
|
||||
config_course_cohorts(self.course, [], cohorted=True,
|
||||
auto_cohort_groups=["AutoGroup1", "AutoGroup2"])
|
||||
|
||||
# Will create cohort1, cohort2, and cohort3. Auto cohorts remain uncreated.
|
||||
self._create_cohorts()
|
||||
# Get the cohorts from the course, which will cause auto cohorts to be created.
|
||||
actual_cohorts = self.request_list_cohorts(self.course)
|
||||
# Get references to the created auto cohorts.
|
||||
auto_cohort_1 = get_cohort_by_name(self.course.id, "AutoGroup1")
|
||||
auto_cohort_2 = get_cohort_by_name(self.course.id, "AutoGroup2")
|
||||
expected_cohorts = [
|
||||
ListCohortsTestCase.create_expected_cohort(self.cohort1, 3, CohortAssignmentType.NONE),
|
||||
ListCohortsTestCase.create_expected_cohort(self.cohort2, 2, CohortAssignmentType.NONE),
|
||||
ListCohortsTestCase.create_expected_cohort(self.cohort3, 2, CohortAssignmentType.NONE),
|
||||
ListCohortsTestCase.create_expected_cohort(auto_cohort_1, 0, CohortAssignmentType.RANDOM),
|
||||
ListCohortsTestCase.create_expected_cohort(auto_cohort_2, 0, CohortAssignmentType.RANDOM),
|
||||
]
|
||||
self.verify_lists_expected_cohorts(expected_cohorts, actual_cohorts)
|
||||
|
||||
def test_default_cohort(self):
|
||||
"""
|
||||
Verify that the default cohort is not created and included in the response until students are assigned to it.
|
||||
"""
|
||||
# verify the default cohort is not created when the course is not cohorted
|
||||
self.verify_lists_expected_cohorts([])
|
||||
|
||||
# create a cohorted course without any auto_cohort_groups
|
||||
config_course_cohorts(self.course, [], cohorted=True)
|
||||
|
||||
# verify the default cohort is not yet created until a user is assigned
|
||||
self.verify_lists_expected_cohorts([])
|
||||
|
||||
# create enrolled users
|
||||
users = [UserFactory() for _ in range(3)]
|
||||
self._enroll_users(users, self.course.id)
|
||||
|
||||
# mimic users accessing the discussion forum
|
||||
for user in users:
|
||||
get_cohort(user, self.course.id)
|
||||
|
||||
# verify the default cohort is automatically created
|
||||
default_cohort = get_cohort_by_name(self.course.id, DEFAULT_COHORT_NAME)
|
||||
actual_cohorts = self.request_list_cohorts(self.course)
|
||||
self.verify_lists_expected_cohorts(
|
||||
[ListCohortsTestCase.create_expected_cohort(default_cohort, len(users), CohortAssignmentType.RANDOM)],
|
||||
actual_cohorts,
|
||||
)
|
||||
|
||||
# set auto_cohort_groups and verify the default cohort is no longer listed as RANDOM
|
||||
config_course_cohorts(self.course, [], cohorted=True, auto_cohort_groups=["AutoGroup"])
|
||||
actual_cohorts = self.request_list_cohorts(self.course)
|
||||
auto_cohort = get_cohort_by_name(self.course.id, "AutoGroup")
|
||||
self.verify_lists_expected_cohorts(
|
||||
[
|
||||
ListCohortsTestCase.create_expected_cohort(default_cohort, len(users), CohortAssignmentType.NONE),
|
||||
ListCohortsTestCase.create_expected_cohort(auto_cohort, 0, CohortAssignmentType.RANDOM),
|
||||
],
|
||||
actual_cohorts,
|
||||
)
|
||||
|
||||
|
||||
class AddCohortTestCase(CohortViewsTestCase):
|
||||
"""
|
||||
Tests the `add_cohort` view.
|
||||
"""
|
||||
def request_add_cohort(self, cohort_name, course):
|
||||
"""
|
||||
Call `add_cohort` and return its response as a dict.
|
||||
"""
|
||||
request = RequestFactory().post("dummy_url", {"name": cohort_name})
|
||||
request.user = self.staff_user
|
||||
response = add_cohort(request, course.id.to_deprecated_string())
|
||||
self.assertEqual(response.status_code, 200)
|
||||
return json.loads(response.content)
|
||||
|
||||
def verify_contains_added_cohort(self, response_dict, cohort_name, expected_error_msg=None):
|
||||
"""
|
||||
Check that `add_cohort`'s response correctly returns the newly added
|
||||
cohort (or error) in the response. Also verify that the cohort was
|
||||
actually created/exists.
|
||||
"""
|
||||
if expected_error_msg is not None:
|
||||
self.assertFalse(response_dict.get("success"))
|
||||
self.assertEqual(
|
||||
response_dict.get("msg"),
|
||||
expected_error_msg
|
||||
)
|
||||
else:
|
||||
self.assertTrue(response_dict.get("success"))
|
||||
self.assertEqual(
|
||||
response_dict.get("cohort").get("name"),
|
||||
cohort_name
|
||||
)
|
||||
self.assertIsNotNone(get_cohort_by_name(self.course.id, cohort_name))
|
||||
|
||||
def test_non_staff(self):
|
||||
"""
|
||||
Verify that non-staff users cannot access add_cohort.
|
||||
"""
|
||||
self._verify_non_staff_cannot_access(add_cohort, "POST", [self.course.id.to_deprecated_string()])
|
||||
|
||||
def test_new_cohort(self):
|
||||
"""
|
||||
Verify that we can add a new cohort.
|
||||
"""
|
||||
cohort_name = "New Cohort"
|
||||
self.verify_contains_added_cohort(
|
||||
self.request_add_cohort(cohort_name, self.course),
|
||||
cohort_name,
|
||||
)
|
||||
|
||||
def test_no_cohort(self):
|
||||
"""
|
||||
Verify that we cannot explicitly add no cohort.
|
||||
"""
|
||||
response_dict = self.request_add_cohort("", self.course)
|
||||
self.assertFalse(response_dict.get("success"))
|
||||
self.assertEqual(response_dict.get("msg"), "No name specified")
|
||||
|
||||
def test_existing_cohort(self):
|
||||
"""
|
||||
Verify that we cannot add a cohort with the same name as an existing
|
||||
cohort.
|
||||
"""
|
||||
self._create_cohorts()
|
||||
cohort_name = self.cohort1.name
|
||||
self.verify_contains_added_cohort(
|
||||
self.request_add_cohort(cohort_name, self.course),
|
||||
cohort_name,
|
||||
expected_error_msg="You cannot create two cohorts with the same name"
|
||||
)
|
||||
|
||||
|
||||
class UsersInCohortTestCase(CohortViewsTestCase):
|
||||
"""
|
||||
Tests the `users_in_cohort` view.
|
||||
"""
|
||||
def request_users_in_cohort(self, cohort, course, requested_page, should_return_bad_request=False):
|
||||
"""
|
||||
Call `users_in_cohort` for a given cohort/requested page, and return
|
||||
its response as a dict. When `should_return_bad_request` is True,
|
||||
verify that the response indicates a bad request.
|
||||
"""
|
||||
request = RequestFactory().get("dummy_url", {"page": requested_page})
|
||||
request.user = self.staff_user
|
||||
response = users_in_cohort(request, course.id.to_deprecated_string(), cohort.id)
|
||||
|
||||
if should_return_bad_request:
|
||||
self.assertEqual(response.status_code, 400)
|
||||
return
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
return json.loads(response.content)
|
||||
|
||||
def verify_users_in_cohort_and_response(self, cohort, response_dict, expected_users, expected_page,
|
||||
expected_num_pages):
|
||||
"""
|
||||
Check that the `users_in_cohort` response contains the expected list of
|
||||
users, page number, and total number of pages for a given cohort. Also
|
||||
verify that those users are actually in the given cohort.
|
||||
"""
|
||||
self.assertTrue(response_dict.get("success"))
|
||||
self.assertEqual(response_dict.get("page"), expected_page)
|
||||
self.assertEqual(response_dict.get("num_pages"), expected_num_pages)
|
||||
|
||||
returned_users = User.objects.filter(username__in=[user.get("username") for user in response_dict.get("users")])
|
||||
self.assertItemsEqual(returned_users, expected_users)
|
||||
self.assertTrue(set(returned_users).issubset(cohort.users.all()))
|
||||
|
||||
def test_non_staff(self):
|
||||
"""
|
||||
Verify that non-staff users cannot access `check_users_in_cohort`.
|
||||
"""
|
||||
cohort = CohortFactory(course_id=self.course.id, users=[])
|
||||
self._verify_non_staff_cannot_access(users_in_cohort, "GET", [self.course.id.to_deprecated_string(), cohort.id])
|
||||
|
||||
def test_no_users(self):
|
||||
"""
|
||||
Verify that we don't get back any users for a cohort with no users.
|
||||
"""
|
||||
cohort = CohortFactory(course_id=self.course.id, users=[])
|
||||
response_dict = self.request_users_in_cohort(cohort, self.course, 1)
|
||||
self.verify_users_in_cohort_and_response(
|
||||
cohort,
|
||||
response_dict,
|
||||
expected_users=[],
|
||||
expected_page=1,
|
||||
expected_num_pages=1
|
||||
)
|
||||
|
||||
def test_few_users(self):
|
||||
"""
|
||||
Verify that we get back all users for a cohort when the cohort has
|
||||
<=100 users.
|
||||
"""
|
||||
users = [UserFactory() for _ in range(5)]
|
||||
cohort = CohortFactory(course_id=self.course.id, users=users)
|
||||
response_dict = self.request_users_in_cohort(cohort, self.course, 1)
|
||||
self.verify_users_in_cohort_and_response(
|
||||
cohort,
|
||||
response_dict,
|
||||
expected_users=users,
|
||||
expected_page=1,
|
||||
expected_num_pages=1
|
||||
)
|
||||
|
||||
def test_many_users(self):
|
||||
"""
|
||||
Verify that pagination works correctly for cohorts with >100 users.
|
||||
"""
|
||||
users = [UserFactory() for _ in range(101)]
|
||||
cohort = CohortFactory(course_id=self.course.id, users=users)
|
||||
response_dict_1 = self.request_users_in_cohort(cohort, self.course, 1)
|
||||
response_dict_2 = self.request_users_in_cohort(cohort, self.course, 2)
|
||||
self.verify_users_in_cohort_and_response(
|
||||
cohort,
|
||||
response_dict_1,
|
||||
expected_users=users[:100],
|
||||
expected_page=1,
|
||||
expected_num_pages=2
|
||||
)
|
||||
self.verify_users_in_cohort_and_response(
|
||||
cohort,
|
||||
response_dict_2,
|
||||
expected_users=users[100:],
|
||||
expected_page=2,
|
||||
expected_num_pages=2
|
||||
)
|
||||
|
||||
def test_out_of_range(self):
|
||||
"""
|
||||
Verify that we get a blank page of users when requesting page 0 or a
|
||||
page greater than the actual number of pages.
|
||||
"""
|
||||
users = [UserFactory() for _ in range(5)]
|
||||
cohort = CohortFactory(course_id=self.course.id, users=users)
|
||||
response = self.request_users_in_cohort(cohort, self.course, 0)
|
||||
self.verify_users_in_cohort_and_response(
|
||||
cohort,
|
||||
response,
|
||||
expected_users=[],
|
||||
expected_page=0,
|
||||
expected_num_pages=1
|
||||
)
|
||||
response = self.request_users_in_cohort(cohort, self.course, 2)
|
||||
self.verify_users_in_cohort_and_response(
|
||||
cohort,
|
||||
response,
|
||||
expected_users=[],
|
||||
expected_page=2,
|
||||
expected_num_pages=1
|
||||
)
|
||||
|
||||
def test_non_positive_page(self):
|
||||
"""
|
||||
Verify that we get a `HttpResponseBadRequest` (bad request) when the
|
||||
page we request isn't a positive integer.
|
||||
"""
|
||||
users = [UserFactory() for _ in range(5)]
|
||||
cohort = CohortFactory(course_id=self.course.id, users=users)
|
||||
self.request_users_in_cohort(cohort, self.course, "invalid", should_return_bad_request=True)
|
||||
self.request_users_in_cohort(cohort, self.course, -1, should_return_bad_request=True)
|
||||
|
||||
|
||||
class AddUsersToCohortTestCase(CohortViewsTestCase):
|
||||
"""
|
||||
Tests the `add_users_to_cohort` view.
|
||||
"""
|
||||
def setUp(self):
|
||||
super(AddUsersToCohortTestCase, self).setUp()
|
||||
self._create_cohorts()
|
||||
|
||||
def request_add_users_to_cohort(self, users_string, cohort, course, should_raise_404=False):
|
||||
"""
|
||||
Call `add_users_to_cohort` for a given cohort, course, and list of
|
||||
users, returning its response as a dict. When `should_raise_404` is
|
||||
True, verify that the request raised a Http404.
|
||||
"""
|
||||
request = RequestFactory().post("dummy_url", {"users": users_string})
|
||||
request.user = self.staff_user
|
||||
if should_raise_404:
|
||||
self.assertRaises(
|
||||
Http404,
|
||||
lambda: add_users_to_cohort(request, course.id.to_deprecated_string(), cohort.id)
|
||||
)
|
||||
else:
|
||||
response = add_users_to_cohort(request, course.id.to_deprecated_string(), cohort.id)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
return json.loads(response.content)
|
||||
|
||||
def verify_added_users_to_cohort(self, response_dict, cohort, course, expected_added, expected_changed,
|
||||
expected_present, expected_unknown):
|
||||
"""
|
||||
Check that add_users_to_cohort returned the expected response and has
|
||||
the expected side effects.
|
||||
|
||||
`expected_added` is a list of users
|
||||
`expected_changed` is a list of (user, previous_cohort) tuples
|
||||
`expected_present` is a list of (user, email/username) tuples where
|
||||
email/username corresponds to the input
|
||||
`expected_unknown` is a list of strings corresponding to the input
|
||||
"""
|
||||
self.assertTrue(response_dict.get("success"))
|
||||
self.assertItemsEqual(
|
||||
response_dict.get("added"),
|
||||
[
|
||||
{"username": user.username, "name": user.profile.name, "email": user.email}
|
||||
for user in expected_added
|
||||
]
|
||||
)
|
||||
self.assertItemsEqual(
|
||||
response_dict.get("changed"),
|
||||
[
|
||||
{
|
||||
"username": user.username,
|
||||
"name": user.profile.name,
|
||||
"email": user.email,
|
||||
"previous_cohort": previous_cohort
|
||||
}
|
||||
for (user, previous_cohort) in expected_changed
|
||||
]
|
||||
)
|
||||
self.assertItemsEqual(
|
||||
response_dict.get("present"),
|
||||
[username_or_email for (_, username_or_email) in expected_present]
|
||||
)
|
||||
self.assertItemsEqual(response_dict.get("unknown"), expected_unknown)
|
||||
for user in expected_added + [user for (user, _) in expected_changed + expected_present]:
|
||||
self.assertEqual(
|
||||
CourseUserGroup.objects.get(
|
||||
course_id=course.id,
|
||||
group_type=CourseUserGroup.COHORT,
|
||||
users__id=user.id
|
||||
),
|
||||
cohort
|
||||
)
|
||||
|
||||
def test_non_staff(self):
|
||||
"""
|
||||
Verify that non-staff users cannot access `check_users_in_cohort`.
|
||||
"""
|
||||
cohort = CohortFactory(course_id=self.course.id, users=[])
|
||||
self._verify_non_staff_cannot_access(
|
||||
add_users_to_cohort,
|
||||
"POST",
|
||||
[self.course.id.to_deprecated_string(), cohort.id]
|
||||
)
|
||||
|
||||
def test_empty(self):
|
||||
"""
|
||||
Verify that adding an empty list of users to a cohort has no result.
|
||||
"""
|
||||
response_dict = self.request_add_users_to_cohort("", self.cohort1, self.course)
|
||||
self.verify_added_users_to_cohort(
|
||||
response_dict,
|
||||
self.cohort1,
|
||||
self.course,
|
||||
expected_added=[],
|
||||
expected_changed=[],
|
||||
expected_present=[],
|
||||
expected_unknown=[]
|
||||
)
|
||||
|
||||
def test_only_added(self):
|
||||
"""
|
||||
Verify that we can add users to their first cohort.
|
||||
"""
|
||||
response_dict = self.request_add_users_to_cohort(
|
||||
",".join([user.username for user in self.cohortless_users]),
|
||||
self.cohort1,
|
||||
self.course
|
||||
)
|
||||
self.verify_added_users_to_cohort(
|
||||
response_dict,
|
||||
self.cohort1,
|
||||
self.course,
|
||||
expected_added=self.cohortless_users,
|
||||
expected_changed=[],
|
||||
expected_present=[],
|
||||
expected_unknown=[]
|
||||
)
|
||||
|
||||
def test_only_changed(self):
|
||||
"""
|
||||
Verify that we can move users to a different cohort.
|
||||
"""
|
||||
response_dict = self.request_add_users_to_cohort(
|
||||
",".join([user.username for user in self.cohort2_users + self.cohort3_users]),
|
||||
self.cohort1,
|
||||
self.course
|
||||
)
|
||||
self.verify_added_users_to_cohort(
|
||||
response_dict,
|
||||
self.cohort1,
|
||||
self.course,
|
||||
expected_added=[],
|
||||
expected_changed=(
|
||||
[(user, self.cohort2.name) for user in self.cohort2_users] +
|
||||
[(user, self.cohort3.name) for user in self.cohort3_users]
|
||||
),
|
||||
expected_present=[],
|
||||
expected_unknown=[]
|
||||
)
|
||||
|
||||
def test_only_present(self):
|
||||
"""
|
||||
Verify that we can 'add' users to their current cohort.
|
||||
"""
|
||||
usernames = [user.username for user in self.cohort1_users]
|
||||
response_dict = self.request_add_users_to_cohort(
|
||||
",".join(usernames),
|
||||
self.cohort1,
|
||||
self.course
|
||||
)
|
||||
self.verify_added_users_to_cohort(
|
||||
response_dict,
|
||||
self.cohort1,
|
||||
self.course,
|
||||
expected_added=[],
|
||||
expected_changed=[],
|
||||
expected_present=[(user, user.username) for user in self.cohort1_users],
|
||||
expected_unknown=[]
|
||||
)
|
||||
|
||||
def test_only_unknown(self):
|
||||
"""
|
||||
Verify that non-existent users are not added.
|
||||
"""
|
||||
usernames = ["unknown_user{}".format(i) for i in range(3)]
|
||||
response_dict = self.request_add_users_to_cohort(
|
||||
",".join(usernames),
|
||||
self.cohort1,
|
||||
self.course
|
||||
)
|
||||
self.verify_added_users_to_cohort(
|
||||
response_dict,
|
||||
self.cohort1,
|
||||
self.course,
|
||||
expected_added=[],
|
||||
expected_changed=[],
|
||||
expected_present=[],
|
||||
expected_unknown=usernames
|
||||
)
|
||||
|
||||
def test_all(self):
|
||||
"""
|
||||
Test all adding conditions together.
|
||||
"""
|
||||
unknowns = ["unknown_user{}".format(i) for i in range(3)]
|
||||
response_dict = self.request_add_users_to_cohort(
|
||||
",".join(
|
||||
unknowns +
|
||||
[
|
||||
user.username
|
||||
for user in self.cohortless_users + self.cohort1_users + self.cohort2_users + self.cohort3_users
|
||||
]
|
||||
),
|
||||
self.cohort1,
|
||||
self.course
|
||||
)
|
||||
self.verify_added_users_to_cohort(
|
||||
response_dict,
|
||||
self.cohort1,
|
||||
self.course,
|
||||
expected_added=self.cohortless_users,
|
||||
expected_changed=(
|
||||
[(user, self.cohort2.name) for user in self.cohort2_users] +
|
||||
[(user, self.cohort3.name) for user in self.cohort3_users]
|
||||
),
|
||||
expected_present=[(user, user.username) for user in self.cohort1_users],
|
||||
expected_unknown=unknowns
|
||||
)
|
||||
|
||||
def test_emails(self):
|
||||
"""
|
||||
Verify that we can use emails to identify users.
|
||||
"""
|
||||
unknown = "unknown_user@example.com"
|
||||
response_dict = self.request_add_users_to_cohort(
|
||||
",".join([
|
||||
self.cohort1_users[0].email,
|
||||
self.cohort2_users[0].email,
|
||||
self.cohortless_users[0].email,
|
||||
unknown
|
||||
]),
|
||||
self.cohort1,
|
||||
self.course
|
||||
)
|
||||
self.verify_added_users_to_cohort(
|
||||
response_dict,
|
||||
self.cohort1,
|
||||
self.course,
|
||||
expected_added=[self.cohortless_users[0]],
|
||||
expected_changed=[(self.cohort2_users[0], self.cohort2.name)],
|
||||
expected_present=[(self.cohort1_users[0], self.cohort1_users[0].email)],
|
||||
expected_unknown=[unknown]
|
||||
)
|
||||
|
||||
def test_delimiters(self):
|
||||
"""
|
||||
Verify that we can use different types of whitespace to delimit
|
||||
usernames in the user string.
|
||||
"""
|
||||
unknown = "unknown_user"
|
||||
response_dict = self.request_add_users_to_cohort(
|
||||
" {} {}\t{}, \r\n{}".format(
|
||||
unknown,
|
||||
self.cohort1_users[0].username,
|
||||
self.cohort2_users[0].username,
|
||||
self.cohortless_users[0].username
|
||||
),
|
||||
self.cohort1,
|
||||
self.course
|
||||
)
|
||||
self.verify_added_users_to_cohort(
|
||||
response_dict,
|
||||
self.cohort1,
|
||||
self.course,
|
||||
expected_added=[self.cohortless_users[0]],
|
||||
expected_changed=[(self.cohort2_users[0], self.cohort2.name)],
|
||||
expected_present=[(self.cohort1_users[0], self.cohort1_users[0].username)],
|
||||
expected_unknown=[unknown]
|
||||
)
|
||||
|
||||
def test_can_cohort_unenrolled_users(self):
|
||||
"""
|
||||
Verify that users can be added to a cohort of a course they're not
|
||||
enrolled in. This feature is currently used to pre-cohort users that
|
||||
are expected to enroll in a course.
|
||||
"""
|
||||
unenrolled_usernames = [user.username for user in self.unenrolled_users]
|
||||
response_dict = self.request_add_users_to_cohort(
|
||||
",".join(unenrolled_usernames),
|
||||
self.cohort1,
|
||||
self.course
|
||||
)
|
||||
self.verify_added_users_to_cohort(
|
||||
response_dict,
|
||||
self.cohort1,
|
||||
self.course,
|
||||
expected_added=self.unenrolled_users,
|
||||
expected_changed=[],
|
||||
expected_present=[],
|
||||
expected_unknown=[]
|
||||
)
|
||||
|
||||
def test_non_existent_cohort(self):
|
||||
"""
|
||||
Verify that an error is raised when trying to add users to a cohort
|
||||
which does not belong to the given course.
|
||||
"""
|
||||
users = [UserFactory(username="user{0}".format(i)) for i in range(3)]
|
||||
usernames = [user.username for user in users]
|
||||
wrong_course_key = SlashSeparatedCourseKey("some", "arbitrary", "course")
|
||||
wrong_course_cohort = CohortFactory(name="wrong_cohort", course_id=wrong_course_key, users=[])
|
||||
self.request_add_users_to_cohort(
|
||||
",".join(usernames),
|
||||
wrong_course_cohort,
|
||||
self.course,
|
||||
should_raise_404=True
|
||||
)
|
||||
|
||||
|
||||
class RemoveUserFromCohortTestCase(CohortViewsTestCase):
|
||||
"""
|
||||
Tests the `remove_user_from_cohort` view.
|
||||
"""
|
||||
def request_remove_user_from_cohort(self, username, cohort):
|
||||
"""
|
||||
Call `remove_user_from_cohort` with the given username and cohort.
|
||||
"""
|
||||
if username is not None:
|
||||
request = RequestFactory().post("dummy_url", {"username": username})
|
||||
else:
|
||||
request = RequestFactory().post("dummy_url")
|
||||
request.user = self.staff_user
|
||||
response = remove_user_from_cohort(request, self.course.id.to_deprecated_string(), cohort.id)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
return json.loads(response.content)
|
||||
|
||||
def verify_removed_user_from_cohort(self, username, response_dict, cohort, expected_error_msg=None):
|
||||
"""
|
||||
Check that `remove_user_from_cohort` properly removes a user from a
|
||||
cohort and returns appropriate success. If the removal should fail,
|
||||
verify that the returned error message matches the expected one.
|
||||
"""
|
||||
if expected_error_msg is None:
|
||||
self.assertTrue(response_dict.get("success"))
|
||||
self.assertIsNone(response_dict.get("msg"))
|
||||
self.assertFalse(self._user_in_cohort(username, cohort))
|
||||
else:
|
||||
self.assertFalse(response_dict.get("success"))
|
||||
self.assertEqual(response_dict.get("msg"), expected_error_msg)
|
||||
|
||||
def test_non_staff(self):
|
||||
"""
|
||||
Verify that non-staff users cannot access `check_users_in_cohort`.
|
||||
"""
|
||||
cohort = CohortFactory(course_id=self.course.id, users=[])
|
||||
self._verify_non_staff_cannot_access(
|
||||
remove_user_from_cohort,
|
||||
"POST",
|
||||
[self.course.id.to_deprecated_string(), cohort.id]
|
||||
)
|
||||
|
||||
def test_no_username_given(self):
|
||||
"""
|
||||
Verify that we get an error message when omitting a username.
|
||||
"""
|
||||
cohort = CohortFactory(course_id=self.course.id, users=[])
|
||||
response_dict = self.request_remove_user_from_cohort(None, cohort)
|
||||
self.verify_removed_user_from_cohort(
|
||||
None,
|
||||
response_dict,
|
||||
cohort,
|
||||
expected_error_msg='No username specified'
|
||||
)
|
||||
|
||||
def test_user_does_not_exist(self):
|
||||
"""
|
||||
Verify that we get an error message when the requested user to remove
|
||||
does not exist.
|
||||
"""
|
||||
username = "bogus"
|
||||
cohort = CohortFactory(course_id=self.course.id, users=[])
|
||||
response_dict = self.request_remove_user_from_cohort(
|
||||
username,
|
||||
cohort
|
||||
)
|
||||
self.verify_removed_user_from_cohort(
|
||||
username,
|
||||
response_dict,
|
||||
cohort,
|
||||
expected_error_msg='No user \'{0}\''.format(username)
|
||||
)
|
||||
|
||||
def test_can_remove_user_not_in_cohort(self):
|
||||
"""
|
||||
Verify that we can "remove" a user from a cohort even if they are not a
|
||||
member of that cohort.
|
||||
"""
|
||||
user = UserFactory()
|
||||
cohort = CohortFactory(course_id=self.course.id, users=[])
|
||||
response_dict = self.request_remove_user_from_cohort(user.username, cohort)
|
||||
self.verify_removed_user_from_cohort(user.username, response_dict, cohort)
|
||||
|
||||
def test_can_remove_user_from_cohort(self):
|
||||
"""
|
||||
Verify that we can remove a user from a cohort.
|
||||
"""
|
||||
user = UserFactory()
|
||||
cohort = CohortFactory(course_id=self.course.id, users=[user])
|
||||
response_dict = self.request_remove_user_from_cohort(user.username, cohort)
|
||||
self.verify_removed_user_from_cohort(user.username, response_dict, cohort)
|
||||
261
openedx/core/djangoapps/course_groups/views.py
Normal file
261
openedx/core/djangoapps/course_groups/views.py
Normal file
@@ -0,0 +1,261 @@
|
||||
from django_future.csrf import ensure_csrf_cookie
|
||||
from django.views.decorators.http import require_POST
|
||||
from django.contrib.auth.models import User
|
||||
from django.core.paginator import Paginator, EmptyPage
|
||||
from django.core.urlresolvers import reverse
|
||||
from django.http import Http404, HttpResponse, HttpResponseBadRequest
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
|
||||
from opaque_keys.edx.locations import SlashSeparatedCourseKey
|
||||
from courseware.courses import get_course_with_access
|
||||
from edxmako.shortcuts import render_to_response
|
||||
|
||||
from . import cohorts
|
||||
from .models import CourseUserGroup
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def json_http_response(data):
|
||||
"""
|
||||
Return an HttpResponse with the data json-serialized and the right content
|
||||
type header.
|
||||
"""
|
||||
return HttpResponse(json.dumps(data), content_type="application/json")
|
||||
|
||||
|
||||
def split_by_comma_and_whitespace(cstr):
|
||||
"""
|
||||
Split a string both by commas and whitespice. Returns a list.
|
||||
"""
|
||||
return re.split(r'[\s,]+', cstr)
|
||||
|
||||
|
||||
@ensure_csrf_cookie
|
||||
def list_cohorts(request, course_key_string):
|
||||
"""
|
||||
Return json dump of dict:
|
||||
|
||||
{'success': True,
|
||||
'cohorts': [{'name': name, 'id': id}, ...]}
|
||||
"""
|
||||
|
||||
# this is a string when we get it here
|
||||
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_key_string)
|
||||
|
||||
course = get_course_with_access(request.user, 'staff', course_key)
|
||||
|
||||
all_cohorts = [
|
||||
{
|
||||
'name': c.name,
|
||||
'id': c.id,
|
||||
'user_count': c.users.count(),
|
||||
'assignment_type': cohorts.CohortAssignmentType.get(c, course)
|
||||
}
|
||||
for c in cohorts.get_course_cohorts(course)
|
||||
]
|
||||
|
||||
return json_http_response({'success': True,
|
||||
'cohorts': all_cohorts})
|
||||
|
||||
|
||||
@ensure_csrf_cookie
|
||||
@require_POST
|
||||
def add_cohort(request, course_key_string):
|
||||
"""
|
||||
Return json of dict:
|
||||
{'success': True,
|
||||
'cohort': {'id': id,
|
||||
'name': name}}
|
||||
|
||||
or
|
||||
|
||||
{'success': False,
|
||||
'msg': error_msg} if there's an error
|
||||
"""
|
||||
# this is a string when we get it here
|
||||
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_key_string)
|
||||
|
||||
get_course_with_access(request.user, 'staff', course_key)
|
||||
|
||||
name = request.POST.get("name")
|
||||
if not name:
|
||||
return json_http_response({'success': False,
|
||||
'msg': "No name specified"})
|
||||
|
||||
try:
|
||||
cohort = cohorts.add_cohort(course_key, name)
|
||||
except ValueError as err:
|
||||
return json_http_response({'success': False,
|
||||
'msg': str(err)})
|
||||
|
||||
return json_http_response({
|
||||
'success': 'True',
|
||||
'cohort': {
|
||||
'id': cohort.id,
|
||||
'name': cohort.name
|
||||
}
|
||||
})
|
||||
|
||||
|
||||
@ensure_csrf_cookie
|
||||
def users_in_cohort(request, course_key_string, cohort_id):
|
||||
"""
|
||||
Return users in the cohort. Show up to 100 per page, and page
|
||||
using the 'page' GET attribute in the call. Format:
|
||||
|
||||
Returns:
|
||||
Json dump of dictionary in the following format:
|
||||
{'success': True,
|
||||
'page': page,
|
||||
'num_pages': paginator.num_pages,
|
||||
'users': [{'username': ..., 'email': ..., 'name': ...}]
|
||||
}
|
||||
"""
|
||||
# this is a string when we get it here
|
||||
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_key_string)
|
||||
|
||||
get_course_with_access(request.user, 'staff', course_key)
|
||||
|
||||
# this will error if called with a non-int cohort_id. That's ok--it
|
||||
# shoudn't happen for valid clients.
|
||||
cohort = cohorts.get_cohort_by_id(course_key, int(cohort_id))
|
||||
|
||||
paginator = Paginator(cohort.users.all(), 100)
|
||||
try:
|
||||
page = int(request.GET.get('page'))
|
||||
except (TypeError, ValueError):
|
||||
# These strings aren't user-facing so don't translate them
|
||||
return HttpResponseBadRequest('Requested page must be numeric')
|
||||
else:
|
||||
if page < 0:
|
||||
return HttpResponseBadRequest('Requested page must be greater than zero')
|
||||
|
||||
try:
|
||||
users = paginator.page(page)
|
||||
except EmptyPage:
|
||||
users = [] # When page > number of pages, return a blank page
|
||||
|
||||
user_info = [{'username': u.username,
|
||||
'email': u.email,
|
||||
'name': '{0} {1}'.format(u.first_name, u.last_name)}
|
||||
for u in users]
|
||||
|
||||
return json_http_response({'success': True,
|
||||
'page': page,
|
||||
'num_pages': paginator.num_pages,
|
||||
'users': user_info})
|
||||
|
||||
|
||||
@ensure_csrf_cookie
|
||||
@require_POST
|
||||
def add_users_to_cohort(request, course_key_string, cohort_id):
|
||||
"""
|
||||
Return json dict of:
|
||||
|
||||
{'success': True,
|
||||
'added': [{'username': ...,
|
||||
'name': ...,
|
||||
'email': ...}, ...],
|
||||
'changed': [{'username': ...,
|
||||
'name': ...,
|
||||
'email': ...,
|
||||
'previous_cohort': ...}, ...],
|
||||
'present': [str1, str2, ...], # already there
|
||||
'unknown': [str1, str2, ...]}
|
||||
|
||||
Raises Http404 if the cohort cannot be found for the given course.
|
||||
"""
|
||||
# this is a string when we get it here
|
||||
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_key_string)
|
||||
get_course_with_access(request.user, 'staff', course_key)
|
||||
|
||||
try:
|
||||
cohort = cohorts.get_cohort_by_id(course_key, cohort_id)
|
||||
except CourseUserGroup.DoesNotExist:
|
||||
raise Http404("Cohort (ID {cohort_id}) not found for {course_key_string}".format(
|
||||
cohort_id=cohort_id,
|
||||
course_key_string=course_key_string
|
||||
))
|
||||
|
||||
users = request.POST.get('users', '')
|
||||
added = []
|
||||
changed = []
|
||||
present = []
|
||||
unknown = []
|
||||
for username_or_email in split_by_comma_and_whitespace(users):
|
||||
if not username_or_email:
|
||||
continue
|
||||
|
||||
try:
|
||||
(user, previous_cohort) = cohorts.add_user_to_cohort(cohort, username_or_email)
|
||||
info = {
|
||||
'username': user.username,
|
||||
'name': user.profile.name,
|
||||
'email': user.email,
|
||||
}
|
||||
if previous_cohort:
|
||||
info['previous_cohort'] = previous_cohort
|
||||
changed.append(info)
|
||||
else:
|
||||
added.append(info)
|
||||
except ValueError:
|
||||
present.append(username_or_email)
|
||||
except User.DoesNotExist:
|
||||
unknown.append(username_or_email)
|
||||
|
||||
return json_http_response({'success': True,
|
||||
'added': added,
|
||||
'changed': changed,
|
||||
'present': present,
|
||||
'unknown': unknown})
|
||||
|
||||
|
||||
@ensure_csrf_cookie
|
||||
@require_POST
|
||||
def remove_user_from_cohort(request, course_key_string, cohort_id):
|
||||
"""
|
||||
Expects 'username': username in POST data.
|
||||
|
||||
Return json dict of:
|
||||
|
||||
{'success': True} or
|
||||
{'success': False,
|
||||
'msg': error_msg}
|
||||
"""
|
||||
# this is a string when we get it here
|
||||
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_key_string)
|
||||
get_course_with_access(request.user, 'staff', course_key)
|
||||
|
||||
username = request.POST.get('username')
|
||||
if username is None:
|
||||
return json_http_response({'success': False,
|
||||
'msg': 'No username specified'})
|
||||
|
||||
cohort = cohorts.get_cohort_by_id(course_key, cohort_id)
|
||||
try:
|
||||
user = User.objects.get(username=username)
|
||||
cohort.users.remove(user)
|
||||
return json_http_response({'success': True})
|
||||
except User.DoesNotExist:
|
||||
log.debug('no user')
|
||||
return json_http_response({'success': False,
|
||||
'msg': "No user '{0}'".format(username)})
|
||||
|
||||
|
||||
def debug_cohort_mgmt(request, course_key_string):
|
||||
"""
|
||||
Debugging view for dev.
|
||||
"""
|
||||
# this is a string when we get it here
|
||||
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_key_string)
|
||||
# add staff check to make sure it's safe if it's accidentally deployed.
|
||||
get_course_with_access(request.user, 'staff', course_key)
|
||||
|
||||
context = {'cohorts_ajax_url': reverse(
|
||||
'cohorts',
|
||||
kwargs={'course_key': course_key.to_deprecated_string()}
|
||||
)}
|
||||
return render_to_response('/course_groups/debug.html', context)
|
||||
@@ -7,7 +7,7 @@ Stores global metadata using the UserPreference model, and per-course metadata u
|
||||
UserCourseTag model.
|
||||
"""
|
||||
|
||||
from user_api.models import UserCourseTag
|
||||
from ..models import UserCourseTag
|
||||
|
||||
# Scopes
|
||||
# (currently only allows per-course tags. Can be expanded to support
|
||||
|
||||
@@ -3,6 +3,7 @@ Test the user api's partition extensions.
|
||||
"""
|
||||
from collections import defaultdict
|
||||
from mock import patch
|
||||
from unittest import TestCase
|
||||
|
||||
from openedx.core.djangoapps.user_api.partition_schemes import RandomUserPartitionScheme, UserPartitionError
|
||||
from student.tests.factories import UserFactory
|
||||
@@ -105,3 +106,15 @@ class TestRandomUserPartitionScheme(PartitionTestCase):
|
||||
# Now, get a new group using the same call
|
||||
new_group = RandomUserPartitionScheme.get_group_for_user(self.MOCK_COURSE_ID, self.user, user_partition)
|
||||
self.assertEqual(old_group.id, new_group.id)
|
||||
|
||||
|
||||
class TestExtension(TestCase):
|
||||
"""
|
||||
Ensure that the scheme extension is correctly plugged in (via entry point
|
||||
in setup.py)
|
||||
"""
|
||||
|
||||
def test_get_scheme(self):
|
||||
self.assertEqual(UserPartition.get_scheme('random'), RandomUserPartitionScheme)
|
||||
with self.assertRaisesRegexp(UserPartitionError, 'Unrecognized scheme'):
|
||||
UserPartition.get_scheme('other')
|
||||
|
||||
@@ -888,7 +888,7 @@ class RegistrationViewTest(ApiTestCase):
|
||||
no_extra_fields_setting = {}
|
||||
|
||||
with simulate_running_pipeline(
|
||||
"user_api.views.third_party_auth.pipeline",
|
||||
"openedx.core.djangoapps.user_api.views.third_party_auth.pipeline",
|
||||
"google-oauth2", email="bob@example.com",
|
||||
fullname="Bob", username="Bob123"
|
||||
):
|
||||
|
||||
@@ -27,7 +27,7 @@ from edxmako.shortcuts import marketing_link
|
||||
from util.authentication import SessionAuthenticationAllowInactiveUser
|
||||
from .api import account as account_api, profile as profile_api
|
||||
from .helpers import FormDescription, shim_student_view, require_post_params
|
||||
from .models import UserPreference
|
||||
from .models import UserPreference, UserProfile
|
||||
from .serializers import UserSerializer, UserPreferenceSerializer
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user