EDUCATOR-5069: Display student key in csv export (#24235)

* use external_user_key in teams csv download

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': <username>,
+            '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>,
             'mode': <student enrollment mode for the given course>,
             <teamset id>: <team name> 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:

lms/djangoapps/teams/tests/test_csv.py

 """ 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)