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

feat: teams csv export split username / external key columns (#29981)

Browse Source 
When doing import, we ran into an issue where there was a learner with an external program key that was the
same as an existing, completely unrelated edX account username. Rather than try to guess which learner we want
or do the lookups backwards, it seemed that splitting these columns to avoid any ambiguity would be the most
straightforward and simple approach
This commit is contained in:
Jansen Kantor
2022-03-01 20:04:32 -05:00
committed by GitHub
parent 3d4cdbcfe9
commit 05b8da5d93
3 changed files with 122 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
lms/djangoapps/teams/csv.py
View File

@@ -9,7 +9,7 @@ from django.contrib.auth.models import User # lint-amnesty, pylint: disable=imp
from django.db.models import Prefetch
from common.djangoapps.student.models import CourseEnrollment
from lms.djangoapps.program_enrollments.models import ProgramCourseEnrollment, ProgramEnrollment
from lms.djangoapps.program_enrollments.models import ProgramCourseEnrollment
from lms.djangoapps.teams.api import (
ORGANIZATION_PROTECTED_MODES,
OrganizationProtectionStatus,
@@ -41,9 +41,9 @@ def load_team_membership_csv(course, response):
def _get_team_membership_csv_headers(course):
"""
Get headers for team membership csv.
['user', 'mode', <teamset_id_1>, ..., ,<teamset_id_n>]
['username', 'external_user_id', 'mode', <teamset_id_1>, ..., ,<teamset_id_n>]
"""
headers = ['user', 'mode']
headers = ['username', 'external_user_id', 'mode']
for teamset in sorted(course.teams_configuration.teamsets, key=lambda ts: ts.teamset_id):
headers.append(teamset.teamset_id)
return headers
@@ -54,8 +54,9 @@ def _lookup_team_membership_data(course):
Returns a list of dicts, in the following form:
[
{
'user': If the user is enrolled in this course as a part of a program,
this will be <external_user_key> if the user has one, otherwise <username>,
'username': <edX User username>
'external_user_id': If the user is enrolled in this course as a part of a program,
this will be <external_user_id> if the user has one, otherwise, blank.
'mode': <student enrollment mode for the given course>,
<teamset id>: <team name> for each teamset in which the given user is on a team
}
@@ -73,7 +74,8 @@ def _lookup_team_membership_data(course):
for course_enrollment in course_enrollments:
# This dict contains all the user's team memberships keyed by teamset
student_row = teamset_memberships_by_user.get(course_enrollment.user, {})
student_row['user'] = _get_displayed_user_identifier(course_enrollment)
student_row['username'] = course_enrollment.user.username
student_row['external_user_id'] = _get_external_user_key(course_enrollment)
student_row['mode'] = course_enrollment.mode
team_membership_data.append(student_row)
return team_membership_data
@@ -101,11 +103,11 @@ def _fetch_course_enrollments_with_related_models(course_id):
).order_by('user__username')
def _get_displayed_user_identifier(course_enrollment):
def _get_external_user_key(course_enrollment):
"""
If a user is enrolled in the course as a part of a program and the program identifies them
with an external_user_key, use that as the value of the 'user' column.
Otherwise, use the user's username.
with an external_user_key, return that value for the 'external_user_key' column.
Otherwise, return None.
"""
program_course_enrollments = course_enrollment.programcourseenrollment_set
if program_course_enrollments.exists():
@@ -114,7 +116,7 @@ def _get_displayed_user_identifier(course_enrollment):
external_user_key = program_course_enrollment.program_enrollment.external_user_key
if external_user_key:
return external_user_key
return course_enrollment.user.username
return None
def _group_teamset_memberships_by_user(course_team_memberships):
@@ -185,7 +187,7 @@ class TeamMembershipImportManager:
if not self.validate_teamsets(csv_reader):
return False
self.teamset_ids = csv_reader.fieldnames[2:]
self.teamset_ids = self.get_teamset_ids_from_reader(csv_reader)
row_dictionaries = []
csv_usernames = set()
@@ -197,7 +199,7 @@ class TeamMembershipImportManager:
for row in csv_reader:
if not self.validate_teams_have_matching_teamsets(row):
return False
username = row['user']
username = row['username']
if not username:
continue
if not self.is_username_unique(username, csv_usernames):
@@ -207,9 +209,9 @@ class TeamMembershipImportManager:
if user is None:
continue
if not self.validate_user_enrollment_is_valid(user, row['mode']):
row['user'] = None
row['user_model'] = None
continue
row['user'] = user
row['user_model'] = user
if not self.validate_user_assignment_to_team_and_teamset(row):
return False
row_dictionaries.append(row)
@@ -246,18 +248,25 @@ class TeamMembershipImportManager:
def validate_header(self, csv_reader):
"""
Validates header row to ensure that it contains at a minimum columns called 'user', 'mode'.
Validates header row to ensure that it contains at a minimum columns called 'username', 'mode'.
Teamset validation is handled separately
"""
header = csv_reader.fieldnames
if 'user' not in header:
self.validation_errors.append("Header must contain column 'user'.")
if 'username' not in header:
self.validation_errors.append("Header must contain column 'username'.")
return False
if 'mode' not in header:
self.validation_errors.append("Header must contain column 'mode'.")
return False
return True
def get_teamset_ids_from_reader(self, csv_reader):
"""
The teamsets currently will be directly after 'mode'
"""
mode_index = csv_reader.fieldnames.index('mode')
return csv_reader.fieldnames[mode_index + 1:]
def validate_teamsets(self, csv_reader):
"""
Validates team set ids. Returns true if there are no errors.
@@ -265,7 +274,7 @@ class TeamMembershipImportManager:
Teamset does not exist
Teamset id is duplicated
"""
teamset_ids = csv_reader.fieldnames[2:]
teamset_ids = self.get_teamset_ids_from_reader(csv_reader)
valid_teamset_ids = {ts.teamset_id for ts in self.course.teams_configuration.teamsets}
dupe_set = set()
@@ -329,7 +338,7 @@ class TeamMembershipImportManager:
[andrew],masters,team1,,team3
[joe],masters,,team2,team3
"""
user = row['user']
user = row['user_model']
for teamset_id in self.teamset_ids:
# See if the user is already on a team in the teamset
if (user.id, teamset_id) in self.existing_course_team_memberships:
@@ -448,21 +457,22 @@ class TeamMembershipImportManager:
Also, if there is no change in user's membership, the input row's team name will be nulled out so that no
action will take place further in the processing chain.
"""
user = row['user_model']
for ts_id in self.teamset_ids:
if row[ts_id] is None:
# remove this student from the teamset
try:
self._remove_user_from_teamset_and_emit_signal(row['user'].id, ts_id, self.course.id)
self._remove_user_from_teamset_and_emit_signal(user.id, ts_id, self.course.id)
except CourseTeamMembership.DoesNotExist:
pass
else:
# reassignment happens only if proposed team membership is different from existing team membership
if (row['user'].id, ts_id) in self.existing_course_team_memberships:
current_user_teams_name = self.existing_course_team_memberships[row['user'].id, ts_id].name
if (user.id, ts_id) in self.existing_course_team_memberships:
current_user_teams_name = self.existing_course_team_memberships[user.id, ts_id].name
if current_user_teams_name != row[ts_id]:
try:
self._remove_user_from_teamset_and_emit_signal(row['user'].id, ts_id, self.course.id)
del self.existing_course_team_memberships[row['user'].id, ts_id]
self._remove_user_from_teamset_and_emit_signal(user.id, ts_id, self.course.id)
del self.existing_course_team_memberships[user.id, ts_id]
except CourseTeamMembership.DoesNotExist:
pass
else:
@@ -508,11 +518,11 @@ class TeamMembershipImportManager:
"""
Creates a CourseTeamMembership entry - i.e: a relationship between a user and a team.
user_row is a dictionary where key is column name and value is the row value.
{'mode': ' masters','topic_0': '','topic_1': 'team 2','topic_2': None,'user': <user_obj>}
{'mode': ' masters','topic_0': '','topic_1': 'team 2','topic_2': None,'user_model': <user_obj>}
andrew,masters,team1,,team3
joe,masters,,team2,team3
"""
user = user_row['user']
user = user_row['user_model']
for teamset_id in self.teamset_ids:
team_name = user_row[teamset_id]
if not team_name:
@@ -541,22 +551,13 @@ class TeamMembershipImportManager:
}
)
def get_user(self, user_name):
def get_user(self, username):
"""
Gets the user object from user_name/email/locator
user_name: the user_name/email/user locator
"""
try:
return User.objects.get(username=user_name)
return User.objects.get(username=username)
except User.DoesNotExist:
try:
return User.objects.get(email=user_name)
except User.DoesNotExist:
try:
user = ProgramEnrollment.objects.get(external_user_key=user_name).user
if user is None:
return None
return user
except ProgramEnrollment.DoesNotExist:
self.validation_errors.append('User name/email/external key: ' + user_name + ' does not exist.')
return None
self.validation_errors.append('User ' + username + ' does not exist.')
return None

108
lms/djangoapps/teams/tests/test_csv.py
View File

@@ -38,7 +38,7 @@ def csv_import(course, csv_dict_rows):
def csv_export(course):
"""
Call csv.load_team_membership_csv for the given course, and return the result.
The result is returned in the form of a dictionary keyed by the 'user' identifiers for each row,
The result is returned in the form of a dictionary keyed by the 'username' identifiers for each row,
mapping to the full parsed dictionary for that row of the csv.
Returns: DictReader for the returned csv file
@@ -50,16 +50,17 @@ def csv_export(course):
def _user_keyed_dict(reader):
""" create a dict of the rows of the csv, keyed by the "user" value """
return {row['user']: row for row in reader}
""" create a dict of the rows of the csv, keyed by the "username" value """
return {row['username']: row for row in reader}
def _csv_dict_row(user, mode, **kwargs):
def _csv_dict_row(username, external_user_id, mode, **kwargs):
"""
Convenience method to create dicts to pass to csv_import
"""
csv_dict_row = dict(kwargs)
csv_dict_row['user'] = user
csv_dict_row['username'] = username
csv_dict_row['external_user_id'] = external_user_id
csv_dict_row['mode'] = mode
return csv_dict_row
@@ -125,28 +126,29 @@ class TeamMembershipCsvTests(SharedModuleStoreTestCase):
def test_get_headers(self):
# pylint: disable=protected-access
headers = csv._get_team_membership_csv_headers(self.course)
assert headers == ['user', 'mode', 'teamset_1', 'teamset_2', 'teamset_3', 'teamset_4']
assert headers == ['username', 'external_user_id', 'mode', 'teamset_1', 'teamset_2', 'teamset_3', 'teamset_4']
def test_get_headers_no_teamsets(self):
# pylint: disable=protected-access
headers = csv._get_team_membership_csv_headers(self.course_no_teamsets)
assert headers == ['user', 'mode']
assert headers == ['username', 'external_user_id', 'mode']
def test_lookup_team_membership_data(self):
with self.assertNumQueries(3):
# pylint: disable=protected-access
data = csv._lookup_team_membership_data(self.course)
assert len(data) == 5
self.assert_teamset_membership(data[0], 'user1', 'audit', 'team_1_1', 'team_2_2', 'team_3_1')
self.assert_teamset_membership(data[1], 'user2', 'verified', 'team_1_1', 'team_2_2', 'team_3_1')
self.assert_teamset_membership(data[2], 'user3', 'honors', None, 'team_2_1', 'team_3_1')
self.assert_teamset_membership(data[3], 'user4', 'masters', None, None, 'team_3_2')
self.assert_teamset_membership(data[4], 'user5', 'masters', None, None, None)
self.assert_teamset_membership(data[0], 'user1', None, 'audit', 'team_1_1', 'team_2_2', 'team_3_1')
self.assert_teamset_membership(data[1], 'user2', None, 'verified', 'team_1_1', 'team_2_2', 'team_3_1')
self.assert_teamset_membership(data[2], 'user3', None, 'honors', None, 'team_2_1', 'team_3_1')
self.assert_teamset_membership(data[3], 'user4', None, 'masters', None, None, 'team_3_2')
self.assert_teamset_membership(data[4], 'user5', None, 'masters', None, None, None)
def assert_teamset_membership(
self,
user_row,
expected_username,
expected_external_user_id,
expected_mode,
expected_teamset_1_team,
expected_teamset_2_team,
@@ -158,17 +160,27 @@ class TeamMembershipCsvTests(SharedModuleStoreTestCase):
-mode
-team name for teamset_(123)
"""
assert user_row['user'] == expected_username
assert user_row['username'] == expected_username
assert user_row['external_user_id'] == expected_external_user_id
assert user_row['mode'] == expected_mode
assert user_row.get('teamset_1') == expected_teamset_1_team
assert user_row.get('teamset_2') == expected_teamset_2_team
assert user_row.get('teamset_3') == expected_teamset_3_team
def test_load_team_membership_csv(self):
expected_csv_headers = ['user', 'mode', 'teamset_1', 'teamset_2', 'teamset_3', 'teamset_4']
expected_csv_headers = [
'username',
'external_user_id',
'mode',
'teamset_1',
'teamset_2',
'teamset_3',
'teamset_4'
]
expected_data = {}
expected_data['user1'] = _csv_dict_row(
'user1',
'',
'audit',
teamset_1='team_1_1',
teamset_2='team_2_2',
@@ -176,19 +188,21 @@ class TeamMembershipCsvTests(SharedModuleStoreTestCase):
)
expected_data['user2'] = _csv_dict_row(
'user2',
'',
'verified',
teamset_1='team_1_1',
teamset_2='team_2_2',
teamset_3='team_3_1',
)
expected_data['user3'] = _csv_dict_row('user3', 'honors', teamset_2='team_2_1', teamset_3='team_3_1')
expected_data['user4'] = _csv_dict_row('user4', 'masters', teamset_3='team_3_2')
expected_data['user5'] = _csv_dict_row('user5', 'masters')
expected_data['user3'] = _csv_dict_row('user3', '', 'honors', teamset_2='team_2_1', teamset_3='team_3_1')
expected_data['user4'] = _csv_dict_row('user4', '', 'masters', teamset_3='team_3_2')
expected_data['user5'] = _csv_dict_row('user5', '', 'masters')
self._add_blanks_to_expected_data(expected_data, expected_csv_headers)
reader = csv_export(self.course)
assert expected_csv_headers == reader.fieldnames
self.assertDictEqual(expected_data, _user_keyed_dict(reader))
actual_data = _user_keyed_dict(reader)
self.assertDictEqual(expected_data, actual_data)
def _add_blanks_to_expected_data(self, expected_data, headers):
""" Helper method to fill in the "blanks" in test data """
@@ -249,13 +263,11 @@ class TeamMembershipImportManagerTests(TeamMembershipEventTestMixin, SharedModul
def test_load_course_teams(self):
"""
Lodaing course teams shold get the users by team with only 2 queries
Loading course teams should get the users by team with only 2 queries
1 for teams, 1 for user count
"""
team1 = CourseTeamFactory.create(course_id=self.course.id) # lint-amnesty, pylint: disable=unused-variable
team2 = CourseTeamFactory.create(course_id=self.course.id) # lint-amnesty, pylint: disable=unused-variable
team3 = CourseTeamFactory.create(course_id=self.course.id) # lint-amnesty, pylint: disable=unused-variable
team4 = CourseTeamFactory.create(course_id=self.course.id) # lint-amnesty, pylint: disable=unused-variable
for _ in range(4):
CourseTeamFactory.create(course_id=self.course.id)
with self.assertNumQueries(2):
self.import_manager.load_course_teams()
@@ -267,7 +279,7 @@ class TeamMembershipImportManagerTests(TeamMembershipEventTestMixin, SharedModul
row = {
'mode': 'masters',
'teamset_1': 'new_protected_team',
'user': masters_learner
'user_model': masters_learner
}
self.import_manager.add_user_to_team(row)
@@ -282,7 +294,7 @@ class TeamMembershipImportManagerTests(TeamMembershipEventTestMixin, SharedModul
row = {
'mode': 'audit',
'teamset_1': 'new_unprotected_team',
'user': audit_learner
'user_model': audit_learner
}
self.import_manager.add_user_to_team(row)
@@ -313,7 +325,7 @@ class TeamMembershipImportManagerTests(TeamMembershipEventTestMixin, SharedModul
row = {
'mode': 'audit',
'teamset_1': None,
'user': audit_learner
'user_model': audit_learner
}
self.import_manager.remove_user_from_team_for_reassignment(row)
@@ -331,7 +343,7 @@ class TeamMembershipImportManagerTests(TeamMembershipEventTestMixin, SharedModul
team_2 = CourseTeamFactory(course_id=self.course.id, name='test_team_2', topic_id='teamset_1')
team_1.add_user(audit_learner)
csv_row = _csv_dict_row(audit_learner, 'audit', teamset_1=team_2.name)
csv_row = _csv_dict_row(audit_learner, '', 'audit', teamset_1=team_2.name)
csv_import(self.course, [csv_row])
assert not CourseTeamMembership.is_user_on_team(audit_learner, team_1)
@@ -361,7 +373,7 @@ class TeamMembershipImportManagerTests(TeamMembershipEventTestMixin, SharedModul
# ... and I try to add members in excess of capacity
csv_data = self._csv_reader_from_array([
['user', 'mode', 'teamset_1'],
['username', 'mode', 'teamset_1'],
['max_size_0', 'audit', ''],
['max_size_2', 'audit', 'team_1'],
['max_size_3', 'audit', 'team_1'],
@@ -390,7 +402,7 @@ class TeamMembershipImportManagerTests(TeamMembershipEventTestMixin, SharedModul
# When I try to remove them from the team
csv_data = self._csv_reader_from_array([
['user', 'mode', 'teamset_1'],
['username', 'mode', 'teamset_1'],
[user.username, mode, ''],
])
result = self.import_manager.set_team_memberships(csv_data) # lint-amnesty, pylint: disable=unused-variable
@@ -410,12 +422,12 @@ class TeamMembershipImportManagerTests(TeamMembershipEventTestMixin, SharedModul
# When a team is already at/near capaciy
for i in range(3):
user = users[i]
row = {'user': user, 'teamset_1': 'team_1', 'mode': 'audit'}
row = {'user_model': user, 'teamset_1': 'team_1', 'mode': 'audit'}
self.import_manager.add_user_to_team(row)
# ... and I try to switch membership (add/remove)
csv_data = self._csv_reader_from_array([
['user', 'mode', 'teamset_1'],
['username', 'mode', 'teamset_1'],
['learner_4', 'audit', 'team_1'],
['learner_0', 'audit', 'team_2'],
])
@@ -443,7 +455,7 @@ class TeamMembershipImportManagerTests(TeamMembershipEventTestMixin, SharedModul
# When I add them to a team that does not exist
assert CourseTeam.objects.all().count() == 0
csv_data = self._csv_reader_from_array([
['user', 'mode', 'teamset_1'],
['username', 'mode', 'teamset_1'],
[user.username, mode, 'new_exciting_team'],
])
result = self.import_manager.set_team_memberships(csv_data) # lint-amnesty, pylint: disable=unused-variable
@@ -465,7 +477,7 @@ class TeamMembershipImportManagerTests(TeamMembershipEventTestMixin, SharedModul
# When I attempt to add them to the same team
assert CourseTeam.objects.all().count() == 0
csv_data = self._csv_reader_from_array([
['user', 'mode', 'teamset_1'],
['username', 'mode', 'teamset_1'],
[verified_learner.username, 'verified', 'new_exciting_team'],
[masters_learner.username, 'masters', 'new_exciting_team']
])
@@ -487,7 +499,7 @@ class TeamMembershipImportManagerTests(TeamMembershipEventTestMixin, SharedModul
# When I attempt to add a student of an incompatible enrollment mode
masters_learner = self._create_and_enroll_test_user('masters_learner', mode='masters')
csv_data = self._csv_reader_from_array([
['user', 'mode', 'teamset_1'],
['username', 'mode', 'teamset_1'],
[masters_learner.username, 'masters', 'unprotected_team']
])
result = self.import_manager.set_team_memberships(csv_data)
@@ -513,7 +525,7 @@ class TeamMembershipImportManagerTests(TeamMembershipEventTestMixin, SharedModul
# When I attempt to add a student of an incompatible enrollment mode
verified_learner = self._create_and_enroll_test_user('verified_learner', mode='verified')
csv_data = self._csv_reader_from_array([
['user', 'mode', 'teamset_1'],
['username', 'mode', 'teamset_1'],
[verified_learner.username, 'verified', 'protected_team']
])
result = self.import_manager.set_team_memberships(csv_data)
@@ -623,7 +635,8 @@ class ExternalKeyCsvTests(TeamMembershipEventTestMixin, SharedModuleStoreTestCas
self.add_user_to_course_program_team(new_user, add_to_team=False, external_user_key=new_ext_key)
self.assert_user_not_on_team(new_user)
csv_import_row = _csv_dict_row(new_ext_key, 'audit', teamset_id=self.team.name)
csv_import_row = _csv_dict_row(new_user.username, new_ext_key, 'audit', teamset_id=self.team.name)
csv_import(self.course, [csv_import_row])
self.assert_user_on_team(new_user)
self.assert_learner_added_emitted(self.team.team_id, new_user.id)
@@ -632,25 +645,28 @@ class ExternalKeyCsvTests(TeamMembershipEventTestMixin, SharedModuleStoreTestCas
with self.assertNumQueries(3):
# pylint: disable=protected-access
data = csv._lookup_team_membership_data(self.course)
self._assert_test_users_on_team(_user_keyed_dict(data))
self._assert_test_users_on_team(_user_keyed_dict(data), None)
def test_get_csv(self):
reader = csv_export(self.course)
self._assert_test_users_on_team(_user_keyed_dict(reader))
self._assert_test_users_on_team(_user_keyed_dict(reader), '')
def _assert_test_users_on_team(self, data):
def _assert_test_users_on_team(self, data, no_external_key_value):
"""
Assert that the four test users should be listed as members of the team,
and user_in_program should be identified by their external_user_key
and user_in_program should be identified by their external_user_key.
no_external_key_value is used because _lookup_team_membership_data returns `None`
to mean there is no external key, but the CsvWriter library writes `None`s as an empty string
"""
assert len(data) == 4
expected_data = {
user_identifier: _csv_dict_row(user_identifier, 'audit', teamset_id=self.team.name)
for user_identifier in [
self.user_no_program.username,
self.user_in_program_no_external_id.username,
self.user_in_program_not_enrolled_through_program.username,
self.external_user_key
username: _csv_dict_row(username, external_id, 'audit', teamset_id=self.team.name)
for username, external_id in [
(self.user_no_program.username, no_external_key_value),
(self.user_in_program_no_external_id.username, no_external_key_value),
(self.user_in_program_not_enrolled_through_program.username, no_external_key_value),
(self.user_in_program.username, self.external_user_key)
]
}
self.assertDictEqual(expected_data, data)

39
lms/djangoapps/teams/tests/test_views.py
View File

@@ -2831,7 +2831,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
def test_create_membership_via_upload(self):
self.create_and_enroll_student(username='a_user')
csv_content = 'user,mode,topic_0\n'
csv_content = 'username,mode,topic_0\n'
csv_content += 'a_user,audit,team wind power'
csv_file = SimpleUploadedFile('test_file.csv', csv_content.encode('utf8'), content_type='text/csv')
self.client.login(username=self.users['course_staff'].username, password=self.users['course_staff'].password)
@@ -2847,7 +2847,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
def test_upload_invalid_teamset(self):
self.create_and_enroll_student(username='a_user')
csv_content = 'user,mode,topic_0_bad\n'
csv_content = 'username,mode,topic_0_bad\n'
csv_content += 'a_user,audit,team wind power'
csv_file = SimpleUploadedFile('test_file.csv', csv_content.encode('utf8'), content_type='text/csv')
self.client.login(username=self.users['course_staff'].username, password=self.users['course_staff'].password)
@@ -2860,7 +2860,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
)
def test_upload_assign_user_twice_to_same_teamset(self):
csv_content = 'user,mode,topic_0\n'
csv_content = 'username,mode,topic_0\n'
csv_content += 'student_enrolled, masters, team wind power'
csv_file = SimpleUploadedFile('test_file.csv', csv_content.encode('utf8'), content_type='text/csv')
self.client.login(username=self.users['course_staff'].username, password=self.users['course_staff'].password)
@@ -2875,7 +2875,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
self.create_and_enroll_student(username='a_user')
self.create_and_enroll_student(username='b_user')
self.create_and_enroll_student(username='c_user')
csv_content = 'user,mode,topic_0,topic_1,topic_2\n'
csv_content = 'username,mode,topic_0,topic_1,topic_2\n'
csv_content += 'a_user,audit,team wind power,team 2\n'
csv_content += 'b_user,audit,,team 2\n'
csv_content += 'c_user,audit,,,team 3'
@@ -2893,7 +2893,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
assert response_text['message'] == '3 learners were affected.'
def test_upload_non_existing_user(self):
csv_content = 'user,mode,topic_0\n'
csv_content = 'username,mode,topic_0\n'
csv_content += 'missing_user, masters, team wind power'
csv_file = SimpleUploadedFile('test_file.csv', csv_content.encode('utf8'), content_type='text/csv')
self.client.login(username=self.users['course_staff'].username, password=self.users['course_staff'].password)
@@ -2919,7 +2919,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
organization_protected=True
)
csv_content = 'user,mode,topic_1,topic_2\n'
csv_content = 'username,mode,topic_1,topic_2\n'
csv_content += 'a_user,masters,{},{}\n'.format(
existing_team_1.name,
existing_team_2.name
@@ -2953,7 +2953,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
def test_upload_invalid_more_teams_than_teamsets(self):
self.create_and_enroll_student(username='a_user')
csv_content = 'user,mode,topic_1\n'
csv_content = 'username,mode,topic_1\n'
csv_content += 'a_user, masters, team wind power, extra1, extra2'
csv_file = SimpleUploadedFile('test_file.csv', csv_content.encode('utf8'), content_type='text/csv')
self.client.login(username=self.users['course_staff'].username, password=self.users['course_staff'].password)
@@ -2967,7 +2967,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
def test_upload_invalid_student_enrollment_mismatch(self):
self.create_and_enroll_student(username='a_user', mode=CourseMode.AUDIT)
csv_content = 'user,mode,topic_1\n'
csv_content = 'username,mode,topic_1\n'
csv_content += 'a_user,masters,team wind power'
csv_file = SimpleUploadedFile('test_file.csv', csv_content.encode('utf8'), content_type='text/csv')
self.client.login(username=self.users['course_staff'].username, password=self.users['course_staff'].password)
@@ -2986,7 +2986,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
self.create_and_enroll_student(username=masters_username_a, mode=CourseMode.MASTERS)
self.create_and_enroll_student(username=masters_username_b, mode=CourseMode.MASTERS)
csv_content = 'user,mode,topic_1\n'
csv_content = 'username,mode,topic_1\n'
csv_content += f'{audit_username},audit,team wind power\n'
csv_content += f'{masters_username_a},masters,team wind power\n'
csv_content += f'{masters_username_b},masters,team wind power\n'
@@ -3003,7 +3003,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
assert response_text['errors'][0] == expected_error
def test_upload_learners_exceed_max_team_size(self):
csv_content = 'user,mode,topic_0,topic_1\n'
csv_content = 'username,mode,topic_0,topic_1\n'
team1 = 'team wind power'
team2 = 'team 2'
for name_enum in enumerate(['a', 'b', 'c', 'd', 'e', 'f', 'g']):
@@ -3029,7 +3029,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
topic_0_id = 'topic_0'
assert CourseTeamMembership.objects.filter(user_id=self.users[username].id, team__topic_id=topic_0_id).exists()
csv_content = f'user,mode,{topic_0_id},topic_1\n'
csv_content = f'username,mode,{topic_0_id},topic_1\n'
csv_content += f'{username},audit'
csv_file = SimpleUploadedFile('test_file.csv', csv_content.encode('utf8'), content_type='text/csv')
self.client.login(username=self.users['course_staff'].username, password=self.users['course_staff'].password)
@@ -3052,7 +3052,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
windpower_team_name = 'team wind power'
assert CourseTeamMembership.objects \
.filter(user_id=self.users[username].id, team__topic_id=topic_0_id, team__name=windpower_team_name).exists()
csv_content = f'user,mode,{topic_0_id}\n'
csv_content = f'username,mode,{topic_0_id}\n'
csv_content += f'{username},audit,{nuclear_team_name}'
csv_file = SimpleUploadedFile('test_file.csv', csv_content.encode('utf8'), content_type='text/csv')
self.client.login(username=self.users['course_staff'].username, password=self.users['course_staff'].password)
@@ -3082,7 +3082,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
topic_0_id = 'topic_0'
nuclear_team_name = 'team wind power'
assert len(CourseTeamMembership.objects.filter(user_id=self.users[username].id, team__topic_id=topic_0_id)) == 1
csv_content = f'user,mode,{topic_0_id}\n'
csv_content = f'username,mode,{topic_0_id}\n'
csv_content += f'{username},audit,{nuclear_team_name}'
csv_file = SimpleUploadedFile('test_file.csv', csv_content.encode('utf8'), content_type='text/csv')
self.client.login(
@@ -3106,8 +3106,8 @@ class TestBulkMembershipManagement(TeamAPITestCase):
def test_create_membership_via_upload_using_external_key(self):
self.create_and_enroll_student(username='a_user', external_key='a_user_external_key')
csv_content = 'user,mode,topic_0\n'
csv_content += 'a_user_external_key,audit,team wind power'
csv_content = 'username,external_user_id,mode,topic_0\n'
csv_content += 'a_user,a_user_external_key,audit,team wind power'
csv_file = SimpleUploadedFile('test_file.csv', csv_content.encode('utf8'), content_type='text/csv')
self.client.login(username=self.users['course_staff'].username, password=self.users['course_staff'].password)
response = self.make_call(
@@ -3120,10 +3120,11 @@ class TestBulkMembershipManagement(TeamAPITestCase):
response_text = json.loads(response.content.decode('utf-8'))
assert response_text['message'] == '1 learners were affected.'
@unittest.skip("This currently won't fail since we're only using username")
def test_create_membership_via_upload_using_external_key_invalid(self):
self.create_and_enroll_student(username='a_user', external_key='a_user_external_key')
csv_content = 'user,mode,topic_0\n'
csv_content += 'a_user_external_key_invalid,audit,team wind power'
csv_content = 'username,external_user_id,mode,topic_0\n'
csv_content += 'a_user,a_user_external_key_invalid,audit,team wind power'
csv_file = SimpleUploadedFile('test_file.csv', csv_content.encode('utf8'), content_type='text/csv')
self.client.login(username=self.users['course_staff'].username, password=self.users['course_staff'].password)
response = self.make_call(
@@ -3137,7 +3138,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
assert response_text['errors'] == ['User name/email/external key: a_user_external_key_invalid does not exist.']
def test_upload_non_ascii(self):
csv_content = 'user,mode,topic_0\n'
csv_content = 'username,mode,topic_0\n'
team_name = '著文企臺個'
user_name = '著著文企臺個文企臺個'
self.create_and_enroll_student(username=user_name)
@@ -3162,7 +3163,7 @@ class TestBulkMembershipManagement(TeamAPITestCase):
masters_a = 'masters_a'
team = self.wind_team
self.create_and_enroll_student(username=masters_a, mode=CourseMode.MASTERS)
csv_content = f'user,mode,{team.topic_id}\n'
csv_content = f'username,mode,{team.topic_id}\n'
csv_content += f'masters_a, masters,{team.name}'
csv_file = SimpleUploadedFile('test_file.csv', csv_content.encode('utf8'), content_type='text/csv')
self.client.login(username=self.users['course_staff'].username, password=self.users['course_staff'].password)