diff --git a/lms/djangoapps/teams/csv.py b/lms/djangoapps/teams/csv.py index a614e686a2..50e8ed4126 100644 --- a/lms/djangoapps/teams/csv.py +++ b/lms/djangoapps/teams/csv.py @@ -6,6 +6,7 @@ import csv from collections import Counter from django.contrib.auth.models import User +from django.db.models import Prefetch from lms.djangoapps.teams.api import ( OrganizationProtectionStatus, @@ -13,7 +14,7 @@ from lms.djangoapps.teams.api import ( ORGANIZATION_PROTECTED_MODES ) from lms.djangoapps.teams.models import CourseTeam, CourseTeamMembership -from lms.djangoapps.program_enrollments.models import ProgramEnrollment +from lms.djangoapps.program_enrollments.models import ProgramCourseEnrollment, ProgramEnrollment from student.models import CourseEnrollment from .utils import emit_team_event @@ -51,29 +52,69 @@ def _lookup_team_membership_data(course): Returns a list of dicts, in the following form: [ { - 'user': , + 'user': If the user is enrolled in this course as a part of a program, + this will be if the user has one, otherwise , 'mode': , : for each teamset in which the given user is on a team } for student in course ] """ - course_students = CourseEnrollment.objects.users_enrolled_in(course.id).order_by('username') - CourseEnrollment.bulk_fetch_enrollment_states(course_students, course.id) - + # Get course enrollments and team memberships for the given course + course_enrollments = _fetch_course_enrollments_with_related_models(course.id) course_team_memberships = CourseTeamMembership.objects.filter( team__course_id=course.id ).select_related('team', 'user').all() teamset_memberships_by_user = _group_teamset_memberships_by_user(course_team_memberships) + team_membership_data = [] - for user in course_students: - student_row = teamset_memberships_by_user.get(user, dict()) - student_row['user'] = user.username - student_row['mode'], _ = CourseEnrollment.enrollment_mode_for_user(user, course.id) + 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, dict()) + student_row['user'] = _get_displayed_user_identifier(course_enrollment) + student_row['mode'] = course_enrollment.mode team_membership_data.append(student_row) return team_membership_data +def _fetch_course_enrollments_with_related_models(course_id): + """ + Look up active course enrollments for this course. Fetch the user. + Fetch the ProgramCourseEnrollment and ProgramEnrollment if any of the CourseEnrollments are associated with + a program enrollment (so we have access to an external_user_id if it exists). + Order by the username of the enrolled user. + + Returns a QuerySet + """ + return CourseEnrollment.objects.filter( + course_id=course_id, + is_active=True + ).prefetch_related( + Prefetch( + 'programcourseenrollment_set', + queryset=ProgramCourseEnrollment.objects.select_related('program_enrollment') + ) + ).select_related( + 'user' + ).order_by('user__username') + + +def _get_displayed_user_identifier(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. + """ + program_course_enrollments = course_enrollment.programcourseenrollment_set + if program_course_enrollments.exists(): + # A user should only have one or zero ProgramCourseEnrollments associated with a given CourseEnrollment + program_course_enrollment = program_course_enrollments.all()[0] + external_user_key = program_course_enrollment.program_enrollment.external_user_key + if external_user_key: + return external_user_key + return course_enrollment.user.username + + def _group_teamset_memberships_by_user(course_team_memberships): """ Parameters: diff --git a/lms/djangoapps/teams/tests/test_csv.py b/lms/djangoapps/teams/tests/test_csv.py index 2d58c2f2c4..91040f25ca 100644 --- a/lms/djangoapps/teams/tests/test_csv.py +++ b/lms/djangoapps/teams/tests/test_csv.py @@ -1,7 +1,8 @@ """ Tests for the functionality in csv """ +from csv import DictWriter, DictReader +from io import BytesIO, StringIO, TextIOWrapper -from io import StringIO - +from lms.djangoapps.program_enrollments.tests.factories import ProgramEnrollmentFactory, ProgramCourseEnrollmentFactory from lms.djangoapps.teams import csv from lms.djangoapps.teams.models import CourseTeam, CourseTeamMembership from lms.djangoapps.teams.tests.factories import CourseTeamFactory @@ -206,3 +207,132 @@ class TeamMembershipImportManagerTests(SharedModuleStoreTestCase): # They are successfully removed from the team self.assertFalse(CourseTeamMembership.is_user_on_team(audit_learner, course_1_team)) + + +class ExternalKeyCsvTests(SharedModuleStoreTestCase): + """ Tests for functionality related to external_user_keys""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.teamset_id = 'teamset_id' + teams_config = TeamsConfig({ + 'team_sets': [ + { + 'id': cls.teamset_id, + 'name': 'teamset_name', + 'description': 'teamset_desc', + } + ] + }) + cls.course = CourseFactory(teams_configuration=teams_config) + # pylint: disable=protected-access + cls.header_fields = csv._get_team_membership_csv_headers(cls.course) + + cls.team = CourseTeamFactory(course_id=cls.course.id, name='team_name', topic_id=cls.teamset_id) + + cls.user_no_program = UserFactory.create() + cls.user_in_program = UserFactory.create() + cls.user_in_program_no_external_id = UserFactory.create() + cls.user_in_program_not_enrolled_through_program = UserFactory.create() + + # user_no_program is only enrolled in the course + cls.add_user_to_course_program_team(cls.user_no_program, enroll_in_program=False) + + # user_in_program is enrolled in the course and the program, with an external_id + cls.external_user_key = 'externalProgramUserId-123' + cls.add_user_to_course_program_team(cls.user_in_program, external_user_key=cls.external_user_key) + + # user_in_program is enrolled in the course and the program, with no external_id + cls.add_user_to_course_program_team(cls.user_in_program_no_external_id) + + # user_in_program_not_enrolled_through_program is enrolled in a program and the course, but they not connected + cls.add_user_to_course_program_team( + cls.user_in_program_not_enrolled_through_program, connect_enrollments=False + ) + + # initialize import manager + cls.import_manager = csv.TeamMembershipImportManager(cls.course) + cls.import_manager.teamset_ids = {ts.teamset_id for ts in cls.course.teamsets} + + @classmethod + def add_user_to_course_program_team( + cls, user, add_to_team=True, enroll_in_program=True, connect_enrollments=True, external_user_key=None + ): + """ + Set up a test user by enrolling them in self.course, and then optionaly: + - enroll them in a program + - link their program and course enrollments + - give their program enrollment an external_user_key + """ + course_enrollment = CourseEnrollmentFactory.create(user=user, course_id=cls.course.id) + if add_to_team: + cls.team.add_user(user) + if enroll_in_program: + program_enrollment = ProgramEnrollmentFactory.create(user=user, external_user_key=external_user_key) + if connect_enrollments: + ProgramCourseEnrollmentFactory.create( + program_enrollment=program_enrollment, course_enrollment=course_enrollment + ) + + def assert_user_on_team(self, user): + self.assertTrue(CourseTeamMembership.is_user_on_team(user, self.team)) + + def assert_user_not_on_team(self, user): + self.assertFalse(CourseTeamMembership.is_user_on_team(user, self.team)) + + def test_add_user_to_team_with_external_key(self): + # Make a new user with an external_user_key who is enrolled in the course and program, with an external_key, + # but is not on the team + new_user = UserFactory.create() + new_ext_key = "another-external-user-id-FKQP12345" + 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) + + with BytesIO() as mock_csv_file: + with TextIOWrapper(mock_csv_file, write_through=True) as text_wrapper: + # Create the fake csv file + csv_writer = DictWriter(text_wrapper, fieldnames=self.header_fields) + csv_writer.writeheader() + # Add the new user to the team via CSV upload, identified by their external key + csv_writer.writerow({'user': new_ext_key, 'mode': 'audit', self.teamset_id: self.team.name}) + # We need to seek to the beginning of the file so the csv import manager can read it + mock_csv_file.seek(0) + #After processing the file, the user should be on the team + self.import_manager.set_team_membership_from_csv(mock_csv_file) + + self.assert_user_on_team(new_user) + + def test_lookup_team_membership_data(self): + with self.assertNumQueries(3): + # pylint: disable=protected-access + data = csv._lookup_team_membership_data(self.course) + self._assert_test_users_on_team(data) + + def test_get_csv(self): + with StringIO() as read_buf: + csv.load_team_membership_csv(self.course, read_buf) + read_buf.seek(0) + reader = DictReader(read_buf) + team_memberships = list(reader) + self._assert_test_users_on_team(team_memberships) + + def _assert_test_users_on_team(self, data): + """ + 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 + """ + self.assertEqual(len(data), 4) + self.assertEqual( + {user_row['user'] for user_row in data}, + { + 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 + } + ) + for user_row in data: + self.assertEqual(len(user_row), 3) + self.assertEqual(user_row['mode'], 'audit') + self.assertEqual(user_row[self.teamset_id], self.team.name)