feat: connect teams with content groups using dynamic partition generator (#33788)

Implements the connection from the teams feature to the content groups feature. This implementation uses the dynamic partition generator extension point to associate content groups with the users that belong to a Team.

This implementation was heavily inspired by the enrollment tracks dynamic partitions.

cms/djangoapps/contentstore/tests/test_course_settings.py

@@ -39,6 +39,7 @@ from openedx.core.djangoapps.discussions.config.waffle import (
     OVERRIDE_DISCUSSION_LEGACY_SETTINGS_FLAG
 )
 from openedx.core.djangoapps.models.course_details import CourseDetails
+from openedx.core.lib.teams_config import TeamsConfig
 from xmodule.fields import Date  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.modulestore.django import modulestore  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.modulestore.tests.factories import CourseFactory  # lint-amnesty, pylint: disable=wrong-import-order
@@ -1752,6 +1753,85 @@ class CourseMetadataEditingTest(CourseTestCase):
         self.request.user = self.user
         set_current_request(self.request)
 
+    def test_team_content_groups_off(self):
+        """
+        Tests that user_partition_id is not added to the model when content groups for teams are off.
+        """
+        course = CourseFactory.create(
+            teams_configuration=TeamsConfig({
+                'max_team_size': 2,
+                'team_sets': [{
+                    'id': 'arbitrary-topic-id',
+                    'name': 'arbitrary-topic-name',
+                    'description': 'arbitrary-topic-desc'
+                }]
+            })
+        )
+        settings_dict = {
+            "teams_configuration": {
+                "value": {
+                    "max_team_size": 2,
+                    "team_sets": [
+                        {
+                            "id": "topic_3_id",
+                            "name": "Topic 3 Name",
+                            "description": "Topic 3 desc"
+                        },
+                    ]
+                }
+            }
+        }
+
+        _, errors, updated_data = CourseMetadata.validate_and_update_from_json(
+            course,
+            settings_dict,
+            user=self.user
+        )
+
+        self.assertEqual(len(errors), 0)
+        for team_set in updated_data["teams_configuration"]["value"]["team_sets"]:
+            self.assertNotIn("user_partition_id", team_set)
+
+    @patch("cms.djangoapps.models.settings.course_metadata.CONTENT_GROUPS_FOR_TEAMS.is_enabled", lambda _: True)
+    def test_team_content_groups_on(self):
+        """
+        Tests that user_partition_id is added to the model when content groups for teams are on.
+        """
+        course = CourseFactory.create(
+            teams_configuration=TeamsConfig({
+                'max_team_size': 2,
+                'team_sets': [{
+                    'id': 'arbitrary-topic-id',
+                    'name': 'arbitrary-topic-name',
+                    'description': 'arbitrary-topic-desc'
+                }]
+            })
+        )
+        settings_dict = {
+            "teams_configuration": {
+                "value": {
+                    "max_team_size": 2,
+                    "team_sets": [
+                        {
+                            "id": "topic_3_id",
+                            "name": "Topic 3 Name",
+                            "description": "Topic 3 desc"
+                        },
+                    ]
+                }
+            }
+        }
+
+        _, errors, updated_data = CourseMetadata.validate_and_update_from_json(
+            course,
+            settings_dict,
+            user=self.user
+        )
+
+        self.assertEqual(len(errors), 0)
+        for team_set in updated_data["teams_configuration"]["value"]["team_sets"]:
+            self.assertIn("user_partition_id", team_set)
+
 
 class CourseGraderUpdatesTest(CourseTestCase):
     """

cms/djangoapps/contentstore/utils.py

@@ -839,6 +839,9 @@ def get_visibility_partition_info(xblock, course=None):
         if len(partition["groups"]) > 1 or any(group["selected"] for group in partition["groups"]):
             selectable_partitions.append(partition)
 
+    team_user_partitions = get_user_partition_info(xblock, schemes=["team"], course=course)
+    selectable_partitions += team_user_partitions
+
     course_key = xblock.scope_ids.usage_id.course_key
     is_library = isinstance(course_key, LibraryLocator)
     if not is_library and ContentTypeGatingConfig.current(course_key=course_key).studio_override_enabled:

cms/djangoapps/models/settings/course_metadata.py

@@ -13,14 +13,17 @@ from django.utils.translation import gettext as _
 from xblock.fields import Scope
 
 from cms.djangoapps.contentstore import toggles
+from common.djangoapps.util.db import MYSQL_MAX_INT, generate_int_id
 from common.djangoapps.xblock_django.models import XBlockStudioConfigurationFlag
 from openedx.core.djangoapps.course_apps.toggles import exams_ida_enabled
 from openedx.core.djangoapps.discussions.config.waffle_utils import legacy_discussion_experience_enabled
-from openedx.core.lib.teams_config import TeamsetType
+from openedx.core.lib.teams_config import CONTENT_GROUPS_FOR_TEAMS, TeamsConfig, TeamsetType
 from openedx.features.course_experience import COURSE_ENABLE_UNENROLLED_ACCESS_FLAG
 from xmodule.course_block import get_available_providers  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.modulestore.django import modulestore  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.modulestore.exceptions import InvalidProctoringProvider  # lint-amnesty, pylint: disable=wrong-import-order
+from xmodule.partitions.partitions import MINIMUM_STATIC_PARTITION_ID
+from xmodule.partitions.partitions_service import get_all_partitions_for_course
 
 LOGGER = logging.getLogger(__name__)
 
@@ -266,6 +269,10 @@ class CourseMetadata:
                 did_validate = False
                 errors.append({'key': key, 'message': err_message, 'model': model})
 
+        teams_config = key_values.get('teams_configuration')
+        if teams_config:
+            key_values['teams_configuration'] = cls.fill_teams_user_partitions_ids(block, teams_config)
+
         team_setting_errors = cls.validate_team_settings(filtered_dict)
         if team_setting_errors:
             errors = errors + team_setting_errors
@@ -282,6 +289,43 @@ class CourseMetadata:
 
         return did_validate, errors, updated_data
 
+    @staticmethod
+    def get_user_partition_id(block, min_partition_id, max_partition_id):
+        """
+        Get a dynamic partition id that is not already in use.
+        """
+        used_partition_ids = {p.id for p in get_all_partitions_for_course(block)}
+        return generate_int_id(
+            min_partition_id,
+            max_partition_id,
+            used_partition_ids,
+        )
+
+    @classmethod
+    def fill_teams_user_partitions_ids(cls, block, teams_config):
+        """
+        Fill the `user_partition_id` in the team settings if it is not set.
+
+        This is used by the Dynamic Team Partition Generator to create the dynamic user partitions
+        based on the team-sets defined in the course.
+        """
+        if not CONTENT_GROUPS_FOR_TEAMS.is_enabled(block.id):
+            return teams_config
+
+        for team_set in teams_config.teamsets:
+            if not team_set.user_partition_id:
+                team_set.user_partition_id = cls.get_user_partition_id(
+                    block,
+                    MINIMUM_STATIC_PARTITION_ID,
+                    MYSQL_MAX_INT,
+                )
+        return TeamsConfig(
+            {
+                **teams_config.cleaned_data,
+                "team_sets": [team_set.cleaned_data for team_set in teams_config.teamsets],
+            }
+        )
+
     @classmethod
     def update_from_dict(cls, key_values, block, user, save=True):
         """
@@ -358,7 +402,14 @@ class CourseMetadata:
         error_list = []
         valid_teamset_types = [TeamsetType.open.value, TeamsetType.public_managed.value,
                                TeamsetType.private_managed.value, TeamsetType.open_managed.value]
-        valid_keys = {'id', 'name', 'description', 'max_team_size', 'type'}
+        valid_keys = {
+            'id',
+            'name',
+            'description',
+            'max_team_size',
+            'type',
+            'user_partition_id',
+        }
         teamset_type = topic_settings.get('type', {})
         if teamset_type:
             if teamset_type not in valid_teamset_types:

lms/djangoapps/course_api/blocks/tests/test_api.py

@@ -226,8 +226,8 @@ class TestGetBlocksQueryCounts(TestGetBlocksQueryCountsBase):
             )
 
     @ddt.data(
-        (ModuleStoreEnum.Type.split, 2, True, 23),
-        (ModuleStoreEnum.Type.split, 2, False, 13),
+        (ModuleStoreEnum.Type.split, 2, True, 24),
+        (ModuleStoreEnum.Type.split, 2, False, 14),
     )
     @ddt.unpack
     def test_query_counts_uncached(self, store_type, expected_mongo_queries, with_storage_backing, num_sql_queries):

lms/djangoapps/grades/tests/test_course_grade_factory.py

@@ -286,7 +286,7 @@ class TestGradeIteration(SharedModuleStoreTestCase):
             else mock_course_grade.return_value
             for student in self.students
         ]
-        with self.assertNumQueries(8):
+        with self.assertNumQueries(11):
             all_course_grades, all_errors = self._course_grades_and_errors_for(self.course, self.students)
         assert {student: str(all_errors[student]) for student in all_errors} == {
             student3: 'Error for student3.',

lms/djangoapps/instructor_task/tests/test_tasks_helper.py

@@ -406,7 +406,7 @@ class TestInstructorGradeReport(InstructorGradeReportTestCase):
 
         with patch('lms.djangoapps.instructor_task.tasks_helper.runner._get_current_task'):
             with check_mongo_calls(2):
-                with self.assertNumQueries(50):
+                with self.assertNumQueries(53):
                     CourseGradeReport.generate(None, None, course.id, {}, 'graded')
 
     def test_inactive_enrollments(self):

lms/djangoapps/teams/team_partition_scheme.py

@@ -0,0 +1,147 @@
+"""
+Provides a UserPartition driver for teams.
+"""
+import logging
+
+from opaque_keys.edx.keys import CourseKey
+
+from lms.djangoapps.courseware.masquerade import (
+    get_course_masquerade,
+    get_masquerading_user_group,
+    is_masquerading_as_specific_student
+)
+from lms.djangoapps.teams.api import get_teams_in_teamset
+from lms.djangoapps.teams.models import CourseTeamMembership
+from openedx.core.lib.teams_config import CONTENT_GROUPS_FOR_TEAMS
+
+from xmodule.partitions.partitions import (  # lint-amnesty, pylint: disable=wrong-import-order
+    Group,
+    UserPartition
+)
+from xmodule.services import TeamsConfigurationService
+
+
+log = logging.getLogger(__name__)
+
+
+class TeamUserPartition(UserPartition):
+    """Extends UserPartition to support dynamic groups pulled from the current
+    course teams.
+    """
+
+    @property
+    def groups(self):
+        """Dynamically generate groups (based on teams) for this partition.
+
+        Returns:
+            list of Group: The groups in this partition.
+        """
+        course_key = CourseKey.from_string(self.parameters["course_id"])
+        if not CONTENT_GROUPS_FOR_TEAMS.is_enabled(course_key):
+            return []
+
+        # Get the team-set for this partition via the partition parameters and then get the teams in that team-set
+        # to create the groups for this partition.
+        team_sets = TeamsConfigurationService().get_teams_configuration(course_key).teamsets
+        team_set_id = self.parameters["team_set_id"]
+        team_set = next((team_set for team_set in team_sets if team_set.teamset_id == team_set_id), None)
+        teams = get_teams_in_teamset(str(course_key), team_set.teamset_id)
+        return [
+            Group(team.id, str(team.name)) for team in teams
+        ]
+
+
+class TeamPartitionScheme:
+    """Uses course team memberships to map learners into partition groups.
+
+    The scheme is only available if the CONTENT_GROUPS_FOR_TEAMS feature flag is enabled.
+
+    This is how it works:
+    - A user partition is created for each team-set in the course with a unused partition ID generated in runtime
+    by using generate_int_id() with min=MINIMUM_STATIC_PARTITION_ID and max=MYSQL_MAX_INT.
+    - A (Content) group is created for each team in the team-set with the database team ID as the group ID,
+    and the team name as the group name.
+    - A user is assigned to a group if they are a member of the team.
+    """
+
+    @classmethod
+    def get_group_for_user(cls, course_key, user, user_partition):
+        """Get the (Content) Group from the specified user partition for the user.
+
+        A user is assigned to the group via their team membership and any mappings from teams to
+        partitions / groups that might exist.
+
+        Args:
+            course_key (CourseKey): The course key.
+            user (User): The user.
+            user_partition (UserPartition): The user partition.
+
+        Returns:
+            Group: The group in the specified user partition
+        """
+        if not CONTENT_GROUPS_FOR_TEAMS.is_enabled(course_key):
+            return None
+
+        # First, check if we have to deal with masquerading.
+        # If the current user is masquerading as a specific student, use the
+        # same logic as normal to return that student's group. If the current
+        # user is masquerading as a generic student in a specific group, then
+        # return that group.
+        if get_course_masquerade(user, course_key) and not is_masquerading_as_specific_student(user, course_key):
+            return get_masquerading_user_group(course_key, user, user_partition)
+
+        # A user cannot belong to more than one team in a team-set by definition, so we can just get the first team.
+        teams = get_teams_in_teamset(str(course_key), user_partition.parameters["team_set_id"])
+        team_ids = [team.team_id for team in teams]
+        user_team = CourseTeamMembership.get_memberships(user.username, [str(course_key)], team_ids).first()
+        if not user_team:
+            return None
+
+        return Group(user_team.team.id, str(user_team.team.name))
+
+    @classmethod
+    def create_user_partition(cls, id, name, description, groups=None, parameters=None, active=True):    # pylint: disable=redefined-builtin, invalid-name, unused-argument
+        """Create a custom UserPartition to support dynamic groups based on teams.
+
+        A Partition has an id, name, scheme, description, parameters, and a list
+        of groups. The id is intended to be unique within the context where these
+        are used. (e.g., for partitions of users within a course, the ids should
+        be unique per-course). The scheme is used to assign users into groups.
+        The parameters field is used to save extra parameters e.g., location of
+        the course ID for this partition scheme.
+
+        Partitions can be marked as inactive by setting the "active" flag to False.
+        Any group access rule referencing inactive partitions will be ignored
+        when performing access checks.
+
+        Args:
+            id (int): The id of the partition.
+            name (str): The name of the partition.
+            description (str): The description of the partition.
+            groups (list of Group): The groups in the partition.
+            parameters (dict): The parameters for the partition.
+            active (bool): Whether the partition is active.
+
+        Returns:
+            TeamUserPartition: The user partition.
+        """
+        course_key = CourseKey.from_string(parameters["course_id"])
+        if not CONTENT_GROUPS_FOR_TEAMS.is_enabled(course_key):
+            return None
+
+        # Team-set used to create partition was created before this feature was
+        # introduced.  In that case, we need to create a new partition with a
+        # new team-set id.
+        if not id:
+            return None
+
+        team_set_partition = TeamUserPartition(
+            id,
+            str(name),
+            str(description),
+            groups,
+            cls,
+            parameters,
+            active=True,
+        )
+        return team_set_partition

lms/djangoapps/teams/tests/test_partition_scheme.py

@@ -0,0 +1,211 @@
+"""
+Test the partitions and partitions services. The partitions tested
+in this file are the following:
+- TeamPartitionScheme
+"""
+from unittest.mock import MagicMock, patch
+
+from common.djangoapps.student.tests.factories import UserFactory
+from lms.djangoapps.teams.tests.factories import CourseTeamFactory
+from lms.djangoapps.teams.team_partition_scheme import TeamPartitionScheme
+from openedx.core.lib.teams_config import create_team_set_partitions_with_course_id
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase  # lint-amnesty, pylint: disable=wrong-import-order
+from xmodule.modulestore.tests.factories import ToyCourseFactory
+from xmodule.modulestore.django import modulestore
+from xmodule.partitions.partitions import Group
+
+
+@patch(
+    "lms.djangoapps.teams.team_partition_scheme.CONTENT_GROUPS_FOR_TEAMS.is_enabled",
+    lambda _: True
+)
+class TestTeamPartitionScheme(ModuleStoreTestCase):
+    """
+    Test the TeamPartitionScheme partition scheme and its related functions.
+    """
+
+    def setUp(self):
+        """
+        Regenerate a course with teams configuration, partition and groups,
+        and a student for each test.
+        """
+        super().setUp()
+        self.course_key = ToyCourseFactory.create().id
+        self.course = modulestore().get_course(self.course_key)
+        self.student = UserFactory.create()
+        self.student.courseenrollment_set.create(course_id=self.course_key, is_active=True)
+        self.team_sets = [
+            MagicMock(name="1st TeamSet", teamset_id=1, user_partition_id=51),
+            MagicMock(name="2nd TeamSet", teamset_id=2, user_partition_id=52),
+        ]
+
+    @patch("lms.djangoapps.teams.team_partition_scheme.TeamsConfigurationService")
+    def test_create_user_partition_with_course_id(self, mock_teams_configuration_service):
+        """
+        Test that create_user_partition returns the correct user partitions for the input data.
+
+        Expected result:
+        - There's a user partition matching the ID given.
+        """
+        mock_teams_configuration_service().get_teams_configuration.return_value.teamsets = self.team_sets
+
+        partition = TeamPartitionScheme.create_user_partition(
+            id=self.team_sets[0].user_partition_id,
+            name=f"Team Group: {self.team_sets[0].name}",
+            description="Partition for segmenting users by team-set",
+            parameters={
+                "course_id": str(self.course_key),
+                "team_set_id": self.team_sets[0].teamset_id,
+            }
+        )
+
+        assert partition.id == self.team_sets[0].user_partition_id
+
+    def test_team_partition_generator(self):
+        """
+        Test that create_team_set_partition returns the correct user partitions for the input data.
+
+        Expected result:
+        - The user partitions are created based on the team sets.
+        """
+        partitions = create_team_set_partitions_with_course_id(self.course_key, self.team_sets)
+
+        assert partitions == [
+            TeamPartitionScheme.create_user_partition(
+                id=self.team_sets[0].user_partition_id,
+                name=f"Team Group: {self.team_sets[0].name}",
+                description="Partition for segmenting users by team-set",
+                parameters={
+                    "course_id": str(self.course_key),
+                    "team_set_id": self.team_sets[0].teamset_id,
+                }
+            ),
+            TeamPartitionScheme.create_user_partition(
+                id=self.team_sets[1].user_partition_id,
+                name=f"Team Group: {self.team_sets[1].name}",
+                description="Partition for segmenting users by team-set",
+                parameters={
+                    "course_id": str(self.course_key),
+                    "team_set_id": self.team_sets[1].teamset_id,
+                }
+            ),
+        ]
+
+    @patch("lms.djangoapps.teams.team_partition_scheme.TeamsConfigurationService")
+    def test_get_partition_groups(self, mock_teams_configuration_service):
+        """
+        Test that the TeamPartitionScheme returns the correct groups for a team set.
+
+        Expected result:
+        - The groups in the partition match the teams in the team set.
+        """
+        mock_teams_configuration_service().get_teams_configuration.return_value.teamsets = self.team_sets
+        team_1 = CourseTeamFactory.create(
+            name="Team 1 in TeamSet",
+            course_id=self.course_key,
+            topic_id=self.team_sets[0].teamset_id,
+        )
+        team_2 = CourseTeamFactory.create(
+            name="Team 2 in TeamSet",
+            course_id=self.course_key,
+            topic_id=self.team_sets[0].teamset_id,
+        )
+        team_partition_scheme = TeamPartitionScheme.create_user_partition(
+            id=self.team_sets[0].user_partition_id,
+            name=f"Team Group: {self.team_sets[0].name}",
+            description="Partition for segmenting users by team-set",
+            parameters={
+                "course_id": str(self.course_key),
+                "team_set_id": self.team_sets[0].teamset_id,
+            }
+        )
+
+        assert team_partition_scheme.groups == [
+            Group(team_1.id, str(team_1.name)),
+            Group(team_2.id, str(team_2.name)),
+        ]
+
+    @patch("lms.djangoapps.teams.team_partition_scheme.TeamsConfigurationService")
+    def test_get_group_for_user(self, mock_teams_configuration_service):
+        """
+        Test that the TeamPartitionScheme returns the correct group for a
+        student in a team when the team is linked to a partition group.
+
+        Expected result:
+        - The group returned matches the team the student is in.
+        """
+        mock_teams_configuration_service().get_teams_configuration.return_value.teamsets = self.team_sets
+        team = CourseTeamFactory.create(
+            name="Team in 1st TeamSet",
+            course_id=self.course_key,
+            topic_id=self.team_sets[0].teamset_id,
+        )
+        team.add_user(self.student)
+        team_partition_scheme = TeamPartitionScheme.create_user_partition(
+            id=self.team_sets[0].user_partition_id,
+            name=f"Team Group: {self.team_sets[0].name}",
+            description="Partition for segmenting users by team-set",
+            parameters={
+                "course_id": str(self.course_key),
+                "team_set_id": self.team_sets[0].teamset_id,
+            }
+        )
+
+        assert TeamPartitionScheme.get_group_for_user(
+            self.course_key, self.student, team_partition_scheme
+        ) == team_partition_scheme.groups[0]
+
+    def test_get_group_for_user_no_team(self):
+        """
+        Test that the TeamPartitionScheme returns None for a student not in a team.
+
+        Expected result:
+        - The group returned is None.
+        """
+        team_partition_scheme = TeamPartitionScheme.create_user_partition(
+            id=51,
+            name="Team Group: 1st TeamSet",
+            description="Partition for segmenting users by team-set",
+            parameters={
+                "course_id": str(self.course_key),
+                "team_set_id": 1,
+            }
+        )
+
+        assert TeamPartitionScheme.get_group_for_user(
+            self.course_key, self.student, team_partition_scheme
+        ) is None
+
+    @patch("lms.djangoapps.teams.team_partition_scheme.get_course_masquerade")
+    @patch("lms.djangoapps.teams.team_partition_scheme.get_masquerading_user_group")
+    @patch("lms.djangoapps.teams.team_partition_scheme.is_masquerading_as_specific_student")
+    def test_group_for_user_masquerading(
+        self,
+        mock_is_masquerading_as_specific_student,
+        mock_get_masquerading_user_group,
+        mock_get_course_masquerade
+    ):
+        """
+        Test that the TeamPartitionScheme calls the masquerading functions when
+        the user is masquerading.
+
+        Expected result:
+        - The masquerading functions are called.
+        """
+        team_partition_scheme = TeamPartitionScheme.create_user_partition(
+            id=51,
+            name="Team Group: 1st TeamSet",
+            description="Partition for segmenting users by team-set",
+            parameters={
+                "course_id": str(self.course_key),
+                "team_set_id": 1,
+            }
+        )
+        mock_get_course_masquerade.return_value = True
+        mock_is_masquerading_as_specific_student.return_value = False
+
+        TeamPartitionScheme.get_group_for_user(
+            self.course_key, self.student, team_partition_scheme
+        )
+
+        mock_get_masquerading_user_group.assert_called_once_with(self.course_key, self.student, team_partition_scheme)

openedx/core/djangoapps/content/learning_sequences/api/outlines.py

@@ -16,6 +16,8 @@ from opaque_keys import OpaqueKey
 from opaque_keys.edx.keys import CourseKey
 from opaque_keys.edx.locator import LibraryLocator
 from openedx.core import types
+from openedx.core.djangoapps.content.learning_sequences.api.processors.team_partition_groups \
+    import TeamPartitionGroupsOutlineProcessor
 
 from ..data import (
     ContentErrorData,
@@ -330,6 +332,7 @@ def _get_user_course_outline_and_processors(course_key: CourseKey,  # lint-amnes
         ('enrollment', EnrollmentOutlineProcessor),
         ('enrollment_track_partitions', EnrollmentTrackPartitionGroupsOutlineProcessor),
         ('cohorts_partitions', CohortPartitionGroupsOutlineProcessor),
+        ('teams_partitions', TeamPartitionGroupsOutlineProcessor),
     ]
 
     # Run each OutlineProcessor in order to figure out what items we have to

openedx/core/djangoapps/content/learning_sequences/api/processors/team_partition_groups.py

@@ -0,0 +1,99 @@
+"""
+Outline processors for applying team user partition groups.
+"""
+import logging
+from datetime import datetime
+from typing import Dict
+
+from opaque_keys.edx.keys import CourseKey
+
+from openedx.core import types
+from openedx.core.djangoapps.content.learning_sequences.api.processors.base import OutlineProcessor
+from openedx.core.lib.teams_config import create_team_set_partitions_with_course_id, CONTENT_GROUPS_FOR_TEAMS
+from xmodule.partitions.partitions import Group
+from xmodule.partitions.partitions_service import get_user_partition_groups
+
+log = logging.getLogger(__name__)
+
+
+class TeamPartitionGroupsOutlineProcessor(OutlineProcessor):
+    """
+    Processor for applying all user partition groups to the course outline.
+
+    This processor is used to remove content from the course outline based on
+    the user's team membership. It is used in the courseware API to remove
+    content from the course outline before it is returned to the client.
+    """
+    def __init__(self, course_key: CourseKey, user: types.User, at_time: datetime):
+        """
+        Attributes:
+            current_user_groups (Dict[str, Group]): The groups to which the user
+                belongs in each partition.
+        """
+        super().__init__(course_key, user, at_time)
+        self.current_user_groups: Dict[str, Group] = {}
+
+    def load_data(self, _) -> None:
+        """
+        Pull team groups for this course and which group the user is in.
+        """
+        if not CONTENT_GROUPS_FOR_TEAMS.is_enabled(self.course_key):
+            return
+
+        user_partitions = create_team_set_partitions_with_course_id(self.course_key)
+        self.current_user_groups = get_user_partition_groups(
+            self.course_key,
+            user_partitions,
+            self.user,
+            partition_dict_key="id",
+        )
+
+    def _is_user_excluded_by_partition_group(self, user_partition_groups):
+        """
+        Is the user part of the group to which the block is restricting content?
+
+        Arguments:
+            user_partition_groups (Dict[int, Set(int)]): Mapping from partition
+                ID to the groups to which the user belongs in that partition.
+
+        Returns:
+            bool: True if the user is excluded from the content, False otherwise.
+            The user is excluded from the content if and only if, for a non-empty
+            partition group, the user is not in any of the groups for that partition.
+        """
+        if not CONTENT_GROUPS_FOR_TEAMS.is_enabled(self.course_key):
+            return False
+
+        if not user_partition_groups:
+            return False
+
+        for partition_id, groups in user_partition_groups.items():
+            if partition_id not in self.current_user_groups:
+                continue
+            if self.current_user_groups[partition_id].id in groups:
+                return False
+
+        return True
+
+    def usage_keys_to_remove(self, full_course_outline):
+        """
+        Content group exclusions remove the content entirely.
+
+        This method returns the usage keys of all content that should be
+        removed from the course outline based on the user's team membership.
+        In this context, a team within a team-set maps to a user partition group.
+        """
+        removed_usage_keys = set()
+        for section in full_course_outline.sections:
+            remove_all_children = False
+            if self._is_user_excluded_by_partition_group(
+                section.user_partition_groups
+            ):
+                removed_usage_keys.add(section.usage_key)
+                remove_all_children = True
+            for seq in section.sequences:
+                if remove_all_children or self._is_user_excluded_by_partition_group(
+                    seq.user_partition_groups
+                ):
+                    removed_usage_keys.add(seq.usage_key)
+        return removed_usage_keys

openedx/core/djangoapps/content/learning_sequences/api/tests/test_outlines.py

@@ -3,7 +3,7 @@ Top level API tests. Tests API public contracts only. Do not import/create/mock
 models for this app.
 """
 from datetime import datetime, timezone
-from unittest.mock import patch
+from unittest.mock import patch, MagicMock
 import unittest
 
 from django.conf import settings
@@ -16,6 +16,11 @@ from opaque_keys.edx.keys import CourseKey
 from opaque_keys.edx.locator import LibraryLocator
 import attr
 import ddt
+from lms.djangoapps.teams.tests.factories import CourseTeamFactory
+from openedx.core.djangoapps.content.learning_sequences.api.processors.team_partition_groups import (
+    TeamPartitionGroupsOutlineProcessor,
+)
+from openedx.core.djangolib.testing.utils import skip_unless_lms
 import pytest
 
 from openedx.core.djangoapps.course_apps.toggles import EXAMS_IDA
@@ -1962,3 +1967,227 @@ class ContentErrorTestCase(CacheIsolationTestCase):
         assert get_content_errors(course_key) == [
             ContentErrorData(message="Content is Hard", usage_key=None),
         ]
+
+
+@patch(
+    "openedx.core.djangoapps.content.learning_sequences.api.processors.team_partition_groups.CONTENT_GROUPS_FOR_TEAMS.is_enabled",  # lint-amnesty, pylint: disable=line-too-long
+    lambda _: True
+)
+@skip_unless_lms
+class TeamPartitionGroupsTestCase(OutlineProcessorTestCase):
+    """Tests for team partitions processor that affects outlines."""
+
+    @classmethod
+    def _create_and_enroll_learner(cls, username, is_staff=False):
+        """
+        Helper function to create the learner based on the username,
+        then enroll the learner into the test course with VERIFIED
+        mode.
+        Returns the created learner
+        """
+        learner = UserFactory.create(
+            username=username, email='{}@example.com'.format(username), is_staff=is_staff
+        )
+        learner.courseenrollment_set.create(course_id=cls.course_key, is_active=True)
+        return learner
+
+    @classmethod
+    def _setup_course_outline_with_sections(
+        cls,
+        course_sections,
+        course_start_date=datetime(2021, 3, 26, tzinfo=timezone.utc)
+    ):
+        """
+        Helper function to update the course outline under test with
+        the course sections passed in.
+        Returns the newly constructed course outline
+        """
+        set_dates_for_course(
+            cls.course_key,
+            [
+                (
+                    cls.course_key.make_usage_key('course', 'course'),
+                    {'start': course_start_date}
+                )
+            ]
+        )
+
+        new_outline = CourseOutlineData(
+            course_key=cls.course_key,
+            title="User Partition Test Course",
+            published_at=course_start_date,
+            published_version="8ebece4b69dd593d82fe2021",
+            sections=course_sections,
+            self_paced=False,
+            days_early_for_beta=None,
+            entrance_exam_id=None,
+            course_visibility=CourseVisibility.PRIVATE,
+        )
+
+        replace_course_outline(new_outline)
+
+        return new_outline
+
+    @classmethod
+    def setUpTestData(cls):
+        """
+        Set up the test case with the necessary data. Which includes:
+        - Two Mocked team sets mirroring Teams Configurations team-sets created in Studio.
+        - Two teams, one in each team set.
+        - A course outline with two sections, each with a sequence that is visible to a different team via partition
+        groups.
+        - A student enrolled in the course and added to one of the teams.
+        """
+        super().setUpTestData()
+        cls.visibility = VisibilityData(
+            hide_from_toc=False,
+            visible_to_staff_only=False
+        )
+        cls.course_key = CourseKey.from_string("course-v1:OpenEdX+Outlines+TeamPartitions")
+        cls.team_sets = [
+            MagicMock(name="1st TeamSet", teamset_id=1, user_partition_id=51),
+            MagicMock(name="2nd TeamSet", teamset_id=2, user_partition_id=52),
+        ]
+        cls.student = cls._create_and_enroll_learner("student")
+        cls.team_1 = CourseTeamFactory.create(
+            name="Team in 1st TeamSet",
+            course_id=cls.course_key,
+            topic_id=cls.team_sets[0].teamset_id,
+        )
+        CourseTeamFactory.create(
+            name="Team in 2nd TeamSet",
+            course_id=cls.course_key,
+            topic_id=cls.team_sets[1].teamset_id,
+        )
+        cls.team_1.add_user(cls.student)
+        team_1_sequence_partition_groups = {
+            cls.team_sets[0].user_partition_id: frozenset([cls.team_sets[0].teamset_id]),
+        }
+        team_2_sequence_partition_groups = {
+            cls.team_sets[1].user_partition_id: frozenset([cls.team_sets[1].teamset_id]),
+        }
+        cls.outline = cls._setup_course_outline_with_sections(
+            [
+                CourseSectionData(
+                    usage_key=cls.course_key.make_usage_key('chapter', '1'),
+                    title="Section 0",
+                    sequences=[
+                        CourseLearningSequenceData(
+                            usage_key=cls.course_key.make_usage_key('subsection', '1'),
+                            title='Subsection 0',
+                            visibility=cls.visibility,
+                            user_partition_groups=team_1_sequence_partition_groups,
+                        ),
+                        CourseLearningSequenceData(
+                            usage_key=cls.course_key.make_usage_key('subsection', '2'),
+                            title='Subsection 1',
+                            visibility=cls.visibility,
+                            user_partition_groups=team_2_sequence_partition_groups,
+                        ),
+                    ]
+                )
+            ]
+        )
+
+    @patch("lms.djangoapps.teams.team_partition_scheme.TeamsConfigurationService")
+    @patch("openedx.core.lib.teams_config._get_team_sets")
+    def test_load_data_in_partition_processor(self, team_sets_mock, team_configuration_service_mock):
+        """
+        Test that the team partition groups processor loads the data correctly for the given user.
+
+        Expected result:
+        - The number of user groups should be equal to the number of teams the user is a part of.
+        - The current user groups should contain user_partition_id as the key and an instance of the Group model
+          as the value with the teams data.
+        - The current user groups should contain the name of the team as the value for the key equal to the
+        user_partition_id.
+        """
+        team_sets_mock.return_value = self.team_sets
+        team_configuration_service_mock.return_value.get_teams_configuration.teamsets = self.team_sets
+        team_partition_groups_processor = TeamPartitionGroupsOutlineProcessor(
+            self.course_key, self.student, datetime.now()
+        )
+
+        team_partition_groups_processor.load_data(self.outline)
+
+        assert len(team_partition_groups_processor.current_user_groups) == 1
+        assert team_partition_groups_processor.current_user_groups.get(
+            self.team_sets[0].user_partition_id
+        ) is not None
+        assert team_partition_groups_processor.current_user_groups.get(
+            self.team_sets[1].user_partition_id
+        ) is None
+        assert team_partition_groups_processor.current_user_groups.get(
+            self.team_sets[0].user_partition_id
+        ).name == self.team_1.name
+
+    @patch("lms.djangoapps.teams.team_partition_scheme.TeamsConfigurationService")
+    @patch("openedx.core.lib.teams_config._get_team_sets")
+    def test_user_not_excluded_by_partition_group(self, team_sets_mock, team_configuration_service_mock):
+        """
+        Test that the team partition groups processor correctly determines if a user is excluded by the partition
+        groups.
+
+        Expected result:
+        - The user should not be excluded by the partition groups meaning the method should return False.
+        """
+        team_sets_mock.return_value = self.team_sets
+        team_configuration_service_mock.return_value.get_teams_configuration.teamsets = self.team_sets
+        team_partition_groups_processor = TeamPartitionGroupsOutlineProcessor(
+            self.course_key, self.student, datetime.now()
+        )
+        sequence_partition_groups = {
+            self.team_sets[0].user_partition_id: frozenset([self.team_sets[0].teamset_id]),
+            self.team_sets[1].user_partition_id: frozenset([self.team_sets[1].teamset_id]),
+            53: frozenset([100]),
+        }
+        team_partition_groups_processor.load_data(self.outline)
+
+        # pylint: disable=protected-access
+        assert not team_partition_groups_processor._is_user_excluded_by_partition_group(
+            sequence_partition_groups
+        )
+        assert not team_partition_groups_processor._is_user_excluded_by_partition_group([])
+
+    @patch("lms.djangoapps.teams.team_partition_scheme.TeamsConfigurationService")
+    @patch("openedx.core.lib.teams_config._get_team_sets")
+    def test_user_excluded_by_partition_group(self, team_sets_mock, team_configuration_service_mock):
+        """
+        Test that the team partition groups processor correctly determines if a user is excluded by the partition
+        groups.
+
+        Expected result:
+        - The user should not be excluded by the partition groups meaning the method should return False.
+        """
+        team_sets_mock.return_value = self.team_sets
+        team_configuration_service_mock.return_value.get_teams_configuration.teamsets = self.team_sets
+        team_partition_groups_processor = TeamPartitionGroupsOutlineProcessor(
+            self.course_key, self.student, datetime.now()
+        )
+        sequence_partition_groups = {
+            self.team_sets[1].user_partition_id: frozenset([self.team_sets[1].teamset_id]),
+            53: frozenset([100]),
+        }
+        team_partition_groups_processor.load_data(self.outline)
+
+        # pylint: disable=protected-access
+        assert team_partition_groups_processor._is_user_excluded_by_partition_group(sequence_partition_groups)
+
+    @patch("lms.djangoapps.teams.team_partition_scheme.TeamsConfigurationService")
+    @patch("openedx.core.lib.teams_config._get_team_sets")
+    def test_usage_keys_removed(self, team_sets_mock, team_configuration_service_mock):
+        """Test that the team partition groups processor correctly determines the usage keys to remove from the outline.
+
+        Expected result:
+        - The method should return the usage keys that are not visible to the user based on the partition groups.
+        """
+        team_sets_mock.return_value = self.team_sets
+        team_configuration_service_mock.return_value.get_teams_configuration.teamsets = self.team_sets
+        team_partition_groups_processor = TeamPartitionGroupsOutlineProcessor(
+            self.course_key, self.student, datetime.now()
+        )
+        team_partition_groups_processor.load_data(self.outline)
+
+        assert team_partition_groups_processor.usage_keys_to_remove(self.outline) == {
+            self.course_key.make_usage_key('subsection', '2')
+        }

openedx/core/lib/teams_config.py

@@ -1,17 +1,37 @@
 """
 Safe configuration wrapper for Course Teams feature.
 """
-
-
+import logging
 import re
 from enum import Enum
 
 from django.utils.functional import cached_property
+from django.utils.translation import gettext_lazy as _
+from openedx.core.djangoapps.waffle_utils import CourseWaffleFlag
+
+from xmodule.partitions.partitions import UserPartition, UserPartitionError
+log = logging.getLogger(__name__)
 
 # "Arbitrarily large" but still limited
 MANAGED_TEAM_MAX_TEAM_SIZE = 200
 # Arbitrarily arbitrary
 DEFAULT_COURSE_RUN_MAX_TEAM_SIZE = 50
+TEAM_SCHEME = "team"
+TEAMS_NAMESPACE = "teams"
+
+# .. toggle_name: course_teams.content_groups_for_teams
+# .. toggle_implementation: CourseWaffleFlag
+# .. toggle_default: False
+# .. toggle_description: This flag enables content groups for teams. Content groups are virtual groupings of learners
+#    who will see a particular set of course content. When this flag is enabled, course authors can create teams and
+#    assign content to each of them. Then, when a learner joins a team, they will see the content that is assigned to
+#    that team's content group. This flag is only relevant for courses that have teams enabled.
+# .. toggle_use_cases: temporary, opt_in
+# .. toggle_target_removal_date: Teak
+# .. toggle_creation_date: 2024-04-01
+CONTENT_GROUPS_FOR_TEAMS = CourseWaffleFlag(
+    f"{TEAMS_NAMESPACE}.content_groups_for_teams", __name__
+)
 
 
 class TeamsConfig:
@@ -190,7 +210,7 @@ class TeamsetConfig:
         """
         Return developer-helpful string.
         """
-        attrs = ['teamset_id', 'name', 'description', 'max_team_size', 'teamset_type']
+        attrs = ['teamset_id', 'name', 'description', 'max_team_size', 'teamset_type', 'user_partition_id']
         return "<{} {}>".format(
             self.__class__.__name__,
             " ".join(
@@ -230,6 +250,7 @@ class TeamsetConfig:
             'description': self.description,
             'max_team_size': self.max_team_size,
             'type': self.teamset_type.value,
+            'user_partition_id': self.user_partition_id,
         }
 
     @cached_property
@@ -287,6 +308,14 @@ class TeamsetConfig:
         """
         return self.teamset_type == TeamsetType.private_managed
 
+    @cached_property
+    def user_partition_id(self):
+        """
+        The ID of the dynamic user partition for this team-set,
+        falling back to None.
+        """
+        return self._data.get('user_partition_id')
+
 
 class TeamsetType(Enum):
     """
@@ -332,3 +361,63 @@ def _clean_max_team_size(value):
     if value < 0:
         return None
     return value
+
+
+def create_team_set_partitions_with_course_id(course_id, team_sets=None):
+    """
+    Create and return the team-set user partitions based only on course_id.
+    If they cannot be created, None is returned.
+    """
+    if not team_sets:
+        team_sets = _get_team_sets(course_id) or {}
+
+    try:
+        team_scheme = UserPartition.get_scheme(TEAM_SCHEME)
+    except UserPartitionError:
+        log.warning(f"No {TEAM_SCHEME} scheme registered, TeamUserPartition will not be created.")
+        return None
+
+    # Get team-sets from course and create user partitions for each team-set
+    # Then get teams from each team-set and create user groups for each team
+    partitions = []
+    for team_set in team_sets:
+        partition = team_scheme.create_user_partition(
+            id=team_set.user_partition_id,
+            name=_("Team Group: {team_set_name}").format(team_set_name=team_set.name),
+            description=_("Partition for segmenting users by team-set"),
+            parameters={
+                "course_id": str(course_id),
+                "team_set_id": team_set.teamset_id,
+            }
+        )
+        if partition:
+            partitions.append(partition)
+
+    return partitions
+
+
+def create_team_set_partition(course):
+    """
+    Get the dynamic enrollment track user partition based on the team-sets of the course.
+    """
+    if not CONTENT_GROUPS_FOR_TEAMS.is_enabled(course.id):
+        return []
+    return create_team_set_partitions_with_course_id(
+        course.id,
+        _get_team_sets(course.id),
+    )
+
+
+def _get_team_sets(course_key):
+    """
+    Get team-sets of the course.
+    """
+    # Avoid ImportError import by importing at this level:
+    # TeamsConfigurationService -> is_masquerading_as_specific_student -> CourseMode -> CourseOverview
+    # Raises: ImportError: cannot import name 'CourseOverview' from partially initialized module
+    from xmodule.services import TeamsConfigurationService
+    team_sets = TeamsConfigurationService().get_teams_configuration(course_key).teamsets
+    if not team_sets:
+        return None
+
+    return team_sets

openedx/core/lib/tests/test_teams_config.py

@@ -83,6 +83,7 @@ class TeamsConfigTests(TestCase):
                 "description": "",
                 "max_team_size": 10,
                 "type": "private_managed",
+                "user_partition_id": None,
             },
             {
                 "id": "bokonism",
@@ -90,6 +91,7 @@ class TeamsConfigTests(TestCase):
                 "description": "Busy busy busy",
                 "max_team_size": 2,
                 "type": "open",
+                "user_partition_id": None,
             },
         ]
     }
@@ -137,6 +139,7 @@ class TeamsConfigTests(TestCase):
                 "description": "",
                 "max_team_size": None,
                 "type": "open",
+                "user_partition_id": None,
             },
         ],
     }
@@ -156,7 +159,16 @@ class TeamsConfigTests(TestCase):
         # teams should be considered enabled, and the "enabled" field should be set to True.
         "enabled": True,
         "max_team_size": DEFAULT_COURSE_RUN_MAX_TEAM_SIZE,
-        "team_sets": [dict(id="test-teamset", name="test", description="test", type="open", max_team_size=None)],
+        "team_sets": [
+            dict(
+                id="test-teamset",
+                name="test",
+                description="test",
+                type="open",
+                max_team_size=None,
+                user_partition_id=None
+            ),
+        ],
     }
 
     @ddt.data(

setup.py

@@ -107,6 +107,7 @@ setup(
             "verification = openedx.core.djangoapps.user_api.partition_schemes:ReturnGroup1PartitionScheme",
             "enrollment_track = openedx.core.djangoapps.verified_track_content.partition_scheme:EnrollmentTrackPartitionScheme",  # lint-amnesty, pylint: disable=line-too-long
             "content_type_gate = openedx.features.content_type_gating.partitions:ContentTypeGatingPartitionScheme",
+            "team = lms.djangoapps.teams.team_partition_scheme:TeamPartitionScheme",
         ],
         "openedx.block_structure_transformer": [
             "library_content = lms.djangoapps.course_blocks.transformers.library_content:ContentLibraryTransformer",
@@ -181,7 +182,8 @@ setup(
         ],
         'openedx.dynamic_partition_generator': [
             'enrollment_track = xmodule.partitions.enrollment_track_partition_generator:create_enrollment_track_partition',  # lint-amnesty, pylint: disable=line-too-long
-            'content_type_gating = openedx.features.content_type_gating.partitions:create_content_gating_partition'
+            'content_type_gating = openedx.features.content_type_gating.partitions:create_content_gating_partition',
+            'team = openedx.core.lib.teams_config:create_team_set_partition',
         ],
         'xblock.v1': XBLOCKS,
         'xblock_asides.v1': XBLOCKS_ASIDES,

xmodule/partitions/partitions_service.py

@@ -82,7 +82,13 @@ def _get_dynamic_partitions(course):
     for generator in dynamic_partition_generators:
         generated_partition = generator(course)
         if generated_partition:
-            generated_partitions.append(generated_partition)
+            # If the generator returns a list of partitions, add them all to the list.
+            # Otherwise, just add the single partition. This is needed for cases where
+            # a single generator can return multiple partitions, such as the TeamUserPartition.
+            if isinstance(generated_partition, list):
+                generated_partitions.extend(generated_partition)
+            else:
+                generated_partitions.append(generated_partition)
 
     return generated_partitions
 

xmodule/tests/test_course_block.py

@@ -320,7 +320,8 @@ class TeamsConfigurationTestCase(unittest.TestCase):
             "description": description,
             "id": topic_id,
             "type": "open",
-            "max_team_size": None
+            "max_team_size": None,
+            "user_partition_id": None,
         }
 
     def test_teams_enabled_new_course(self):