feat: Enrollment Tracks OutlineProcessor (MST-685)

NOTE: This will require a forced backfill of course outlines to update
the course content data in learning_sequences:

  python manage.py cms backfill_course_outlines --force

Without this backfill, the learning_sequences API will continue to serve
stale content data that has no user partition group data. It won't cause
errors, but it won't do the exclusions properly.

Commit summary:

* Created EnrollmentTrackPartitionGroupsOutlineProcessor to process the
  enrollment_track User Partition Group, allowing Sequences and Sections
  to be removed based on their group_access settings.
* Added user_partition_groups attribute to CourseLearningSequenceData
  and CourseSectionData in learning_sequences/data.py, along with
  backing model data.
* get_outline_from_modulestore now extracts group_access settings from
  Sections and Sequences. It also bubbles up group_access settings from
  Units, meaning that if a Sequence with no group_access setting has
  Units that are all set to show only to the Verified enrollment track,
  then the Sequence will only show to the Verified enrollment track.

This commit adds model-level support for all user partition groups by
capturing all the content group associations (group_access), but it only
implements the code checks for the enrollment track partition. It's not
clear that we want to generalize, since there's only one other partition
type (A/B testing) that is applicable at the outline level.

It's important to note that there is no way to set the group_access for
a Section or Sequence in Studio today. It's only possible by direct
editing of the OLX for import. That being said, the block structures
framework supports applying course groups at this level, and this commit
moves learning_sequences closer to feature parity.

The bubbling up from Units to the parent Sequence was done to mitigate
confusion when a Sequence is entirely composed of Units that are not
visible to the user because of content group restrictions. It's not
clear whether this is something we want to do in the long term, since it
would simplify the code to always specify group_access at the Sequence
level. This first pass is done partially to collect better data about
places in our courses where this kind of usage is already happening.

Most of the EnrollmentTrackPartitionGroupsOutlineProcessor code and its
tests were written by @schenedx.

cms/djangoapps/contentstore/outlines.py

@@ -79,6 +79,114 @@ def _error_for_not_sequence(section, not_sequence):
     )
 
 
+def _bubbled_up_groups_from_units(group_access_from_units):
+    """
+    Return {user_partition_id: [group_ids]} to bubble up from Units to Sequence.
+
+    This is to handle a special case: If *all* of the Units in a sequence have
+    the exact same group for a given user partition, bubble that value up to the
+    Sequence as a whole. For example, say that every Unit in a Sequence has a
+    group_access that looks like: { ENROLLMENT: [MASTERS] } (where both
+    constants are ints). In this case, an Audit user has nothing to see in the
+    Sequence at all, and it's not useful to give them an empty shell. So we'll
+    act as if the Sequence as a whole had that group setting. Note that there is
+    currently no way to set the group_access setting at the sequence level in
+    Studio, so course teams can only manipulate it for individual Units.
+    """
+    # If there are no Units, there's nothing to bubble up.
+    if not group_access_from_units:
+        return {}
+
+    def _normalize_group_access_dict(group_access):
+        return {
+            user_partition_id: sorted(group_ids)  # sorted for easier comparison
+            for user_partition_id, group_ids in group_access.items()
+        }
+
+    normalized_group_access_dicts = [
+        _normalize_group_access_dict(group_access) for group_access in group_access_from_units
+    ]
+    first_unit_group_access = normalized_group_access_dicts[0]
+    rest_of_seq_group_access_list = normalized_group_access_dicts[1:]
+
+    # If there's only a single Unit, bubble up its group_access.
+    if not rest_of_seq_group_access_list:
+        return first_unit_group_access
+
+    # Otherwise, go through the user partitions and groups in our first unit
+    # and compare them to all the other group_access dicts from the units in the
+    # rest of the sequence. Only keep the ones that match exactly and do not
+    # have empty groups.
+    common_group_access = {
+        user_partition_id: group_ids
+        for user_partition_id, group_ids in first_unit_group_access.items()
+        if group_ids and all(
+            group_ids == group_access.get(user_partition_id)
+            for group_access in rest_of_seq_group_access_list
+        )
+    }
+    return common_group_access
+
+
+def _make_user_partition_groups(usage_key, group_access):
+    """
+    Return a (Dict, Optional[ContentErrorData]) of user partition groups.
+
+    The Dict is a mapping of user partition ID to list of group IDs. If any
+    empty groups are encountered, we create a ContentErrorData about that. If
+    there are no empty groups, we pass back (Dict, None).
+    """
+    empty_partitions = sorted(
+        part_id for part_id, group_ids in group_access.items() if not group_ids
+    )
+    empty_partitions_txt = ", ".join([str(part_id) for part_id in empty_partitions])
+    if empty_partitions:
+        error = ContentErrorData(
+            message=(
+                f'<{usage_key.block_type}> with url_name="{usage_key.block_id}"'
+                f' has the following empty group_access user partitions: '
+                f'{empty_partitions_txt}. This would make this content '
+                f'unavailable to anyone except course staff. Ignoring these '
+                f'group_access settings when building outline.'
+            ),
+            usage_key=_remove_version_info(usage_key),
+        )
+    else:
+        error = None
+
+    user_partition_groups = {
+        part_id: group_ids for part_id, group_ids in group_access.items() if group_ids
+    }
+    return user_partition_groups, error
+
+
+def _make_bubbled_up_error(seq_usage_key, user_partition_id, group_ids):
+    return ContentErrorData(
+        message=(
+            f'<{seq_usage_key.block_type}> with url_name="{seq_usage_key.block_id}"'
+            f' was assigned group_ids {group_ids} for user_partition_id '
+            f'{user_partition_id} because all of its child Units had that '
+            f'group_access setting. This is permitted, but is an unusual usage '
+            f'that may cause unexpected behavior while browsing the course.'
+        ),
+        usage_key=_remove_version_info(seq_usage_key),
+    )
+
+
+def _make_not_bubbled_up_error(seq_usage_key, seq_group_access, user_partition_id, group_ids):
+    return ContentErrorData(
+        message=(
+            f'<{seq_usage_key.block_type}> with url_name="{seq_usage_key.block_id}" '
+            f'has children with only group_ids {group_ids} for user_partition_id '
+            f'{user_partition_id}, but its own group_access setting is '
+            f'{seq_group_access}, which takes precedence. This is permitted, '
+            f'but probably not intended, since it means that the content is '
+            f'effectively unusable by anyone except staff.'
+        ),
+        usage_key=_remove_version_info(seq_usage_key),
+    )
+
+
 def _make_section_data(section):
     """
     Return a (CourseSectionData, List[ContentDataError]) from a SectionBlock.
@@ -105,6 +213,13 @@ def _make_section_data(section):
         section_errors.append(_error_for_not_section(section))
         return (None, section_errors)
 
+    section_user_partition_groups, error = _make_user_partition_groups(
+        section.location, section.group_access
+    )
+    # Invalid user partition errors aren't fatal. Just log and continue on.
+    if error:
+        section_errors.append(error)
+
     # We haven't officially killed off problemset and videosequence yet, so
     # treat them as equivalent to sequential for now.
     valid_sequence_tags = ['sequential', 'problemset', 'videosequence']
@@ -115,6 +230,33 @@ def _make_section_data(section):
             section_errors.append(_error_for_not_sequence(section, sequence))
             continue
 
+        seq_user_partition_groups, error = _make_user_partition_groups(
+            sequence.location, sequence.group_access
+        )
+        if error:
+            section_errors.append(error)
+
+        # Bubble up User Partition Group settings from Units if appropriate.
+        sequence_upg_from_units = _bubbled_up_groups_from_units(
+            [unit.group_access for unit in sequence.get_children()]
+        )
+        for user_partition_id, group_ids in sequence_upg_from_units.items():
+            # If there's an existing user partition ID set at the sequence
+            # level, we respect it, even if it seems nonsensical. The hack of
+            # bubbling things up from the Unit level is only done if there's
+            # no conflicting value set at the Sequence level.
+            if user_partition_id not in seq_user_partition_groups:
+                section_errors.append(
+                    _make_bubbled_up_error(sequence.location, user_partition_id, group_ids)
+                )
+                seq_user_partition_groups[user_partition_id] = group_ids
+            else:
+                section_errors.append(
+                    _make_not_bubbled_up_error(
+                        sequence.location, sequence.group_access, user_partition_id, group_ids
+                    )
+                )
+
         sequences_data.append(
             CourseLearningSequenceData(
                 usage_key=_remove_version_info(sequence.location),
@@ -129,6 +271,7 @@ def _make_section_data(section):
                     hide_from_toc=sequence.hide_from_toc,
                     visible_to_staff_only=sequence.visible_to_staff_only,
                 ),
+                user_partition_groups=seq_user_partition_groups,
             )
         )
 
@@ -140,6 +283,7 @@ def _make_section_data(section):
             hide_from_toc=section.hide_from_toc,
             visible_to_staff_only=section.visible_to_staff_only,
         ),
+        user_partition_groups=section_user_partition_groups,
     )
     return section_data, section_errors
 
@@ -158,7 +302,8 @@ def get_outline_from_modulestore(course_key) -> Tuple[CourseOutlineData, List[Co
     content_errors = []
 
     with store.branch_setting(ModuleStoreEnum.Branch.published_only, course_key):
-        course = store.get_course(course_key, depth=2)
+        # Pull course with depth=3 so we prefetch Section -> Sequence -> Unit
+        course = store.get_course(course_key, depth=3)
         sections_data = []
         for section in course.get_children():
             section_data, section_errors = _make_section_data(section)

cms/djangoapps/contentstore/tests/test_outlines.py

@@ -273,6 +273,152 @@ class OutlineFromModuleStoreTestCase(ModuleStoreTestCase):
         assert outline.sections[0].title == section.url_name
         assert outline.sections[0].sequences[0].title == sequence.url_name
 
+    def test_empty_user_partition_groups(self):
+        """
+        Ignore user partition setting if no groups are associated.
+
+        If we didn't ignore it, we would be creating content that can never be
+        seen by any student.
+        """
+        with self.store.bulk_operations(self.course_key):
+            section = ItemFactory.create(
+                parent_location=self.draft_course.location,
+                category='chapter',
+                display_name='Ch 1',
+                group_access={
+                    49: [],
+                    50: [1, 2],
+                    51: [],
+                }
+            )
+            ItemFactory.create(
+                parent_location=section.location,
+                category='sequential',
+                display_name='Seq 1',
+                group_access={
+                    49: [],
+                }
+            )
+
+        outline, errs = get_outline_from_modulestore(self.course_key)
+        assert len(outline.sections) == 1
+        assert len(outline.sequences) == 1
+        assert outline.sections[0].user_partition_groups == {50: [1, 2]}
+        assert outline.sections[0].sequences[0].user_partition_groups == {}
+        assert len(errs) == 2
+
+    def test_bubbled_up_user_partition_groups_no_children(self):
+        """Testing empty case to make sure bubble-up code doesn't break."""
+        with self.store.bulk_operations(self.course_key):
+            section = ItemFactory.create(
+                parent_location=self.draft_course.location,
+                category='chapter',
+                display_name='Ch 0',
+            )
+
+            # Bubble up with no children (nothing happens)
+            ItemFactory.create(
+                parent_location=section.location,
+                category='sequential',
+                display_name='Seq 0',
+                group_access={}
+            )
+
+        outline, _errs = get_outline_from_modulestore(self.course_key)
+        seq_data = outline.sections[0].sequences[0]
+        assert seq_data.user_partition_groups == {}
+
+    def test_bubbled_up_user_partition_groups_one_child(self):
+        """Group settings should bubble up from Unit to Seq. if only one unit"""
+        with self.store.bulk_operations(self.course_key):
+            section = ItemFactory.create(
+                parent_location=self.draft_course.location,
+                category='chapter',
+                display_name='Ch 0',
+            )
+
+            # Bubble up with 1 child (grabs the setting from child)
+            seq_1 = ItemFactory.create(
+                parent_location=section.location,
+                category='sequential',
+                display_name='Seq 1',
+                group_access={}
+            )
+            ItemFactory.create(
+                parent_location=seq_1.location,
+                category='vertical',
+                display_name='Single Vertical',
+                group_access={50: [1, 2]},
+            )
+
+        outline, errs = get_outline_from_modulestore(self.course_key)
+        seq_data = outline.sections[0].sequences[0]
+        assert seq_data.user_partition_groups == {50: [1, 2]}
+        assert len(errs) == 1
+
+    def test_bubbled_up_user_partition_groups_multiple_children(self):
+        """If all Units have the same group_access, bubble up to Sequence."""
+        with self.store.bulk_operations(self.course_key):
+            section = ItemFactory.create(
+                parent_location=self.draft_course.location,
+                category='chapter',
+                display_name='Ch 0',
+            )
+
+            # Bubble up with n children, all matching for one group
+            seq_n = ItemFactory.create(
+                parent_location=section.location,
+                category='sequential',
+                display_name='Seq N',
+                group_access={}
+            )
+            for i in range(4):
+                ItemFactory.create(
+                    parent_location=seq_n.location,
+                    category='vertical',
+                    display_name=f'vertical {i}',
+                    group_access={50: [3, 4], 51: [i]}  # Only 50 should get bubbled up
+                )
+            ItemFactory.create(
+                parent_location=seq_n.location,
+                category='vertical',
+                display_name='vertical 5',
+                group_access={50: [4, 3], 51: [5]}  # Ordering should be normalized
+            )
+
+        outline, errs = get_outline_from_modulestore(self.course_key)
+        seq_data = outline.sections[0].sequences[0]
+        assert seq_data.user_partition_groups == {50: [3, 4]}
+        assert len(errs) == 1
+
+    def test_not_bubbled_up(self):
+        """Don't bubble up from Unit if Seq has a conflicting group_access."""
+        with self.store.bulk_operations(self.course_key):
+            section = ItemFactory.create(
+                parent_location=self.draft_course.location,
+                category='chapter',
+                display_name='Ch 0',
+            )
+
+            # Bubble up with 1 child (grabs the setting from child)
+            seq_1 = ItemFactory.create(
+                parent_location=section.location,
+                category='sequential',
+                display_name='Seq 1',
+                group_access={50: [3, 4]}
+            )
+            ItemFactory.create(
+                parent_location=seq_1.location,
+                category='vertical',
+                display_name='Single Vertical',
+                group_access={50: [1, 2]},
+            )
+
+        outline, errs = get_outline_from_modulestore(self.course_key)
+        seq_data = outline.sections[0].sequences[0]
+        assert seq_data.user_partition_groups == {50: [3, 4]}  # Kept the seq settings
+        assert len(errs) == 1
+
     def _outline_seq_data(self, modulestore_seq):
         """
         (CourseLearningSequenceData, UsageKey) for a Modulestore sequence.

common/lib/xmodule/xmodule/partitions/enrollment_track_partition_generator.py

@@ -19,9 +19,9 @@ log = logging.getLogger(__name__)
 FEATURES = getattr(settings, 'FEATURES', {})
 
 
-def create_enrollment_track_partition(course):
+def create_enrollment_track_partition_with_course_id(course_id):
     """
-    Create and return the dynamic enrollment track user partition.
+    Create and return the dynamic enrollment track user partition based only on course_id.
     If it cannot be created, None is returned.
     """
     if not FEATURES.get('ENABLE_ENROLLMENT_TRACK_USER_PARTITION'):
@@ -33,6 +33,21 @@ def create_enrollment_track_partition(course):
         log.warning("No 'enrollment_track' scheme registered, EnrollmentTrackUserPartition will not be created.")
         return None
 
+    partition = enrollment_track_scheme.create_user_partition(
+        id=ENROLLMENT_TRACK_PARTITION_ID,
+        name=_("Enrollment Track Groups"),
+        description=_("Partition for segmenting users by enrollment track"),
+        parameters={"course_id": str(course_id)}
+    )
+    return partition
+
+
+def create_enrollment_track_partition(course):
+    """
+    Create and return the dynamic enrollment track user partition.
+    If it cannot be created, None is returned.
+    """
+
     used_ids = {p.id for p in course.user_partitions}
     if ENROLLMENT_TRACK_PARTITION_ID in used_ids:
         log.warning(
@@ -44,10 +59,4 @@ def create_enrollment_track_partition(course):
         )
         return None
 
-    partition = enrollment_track_scheme.create_user_partition(
-        id=ENROLLMENT_TRACK_PARTITION_ID,
-        name=_("Enrollment Track Groups"),
-        description=_("Partition for segmenting users by enrollment track"),
-        parameters={"course_id": str(course.id)}
-    )
-    return partition
+    return create_enrollment_track_partition_with_course_id(course.id)

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

@@ -6,15 +6,15 @@ __init__.py imports from here, and is a more stable place to import from.
 import logging
 from collections import defaultdict
 from datetime import datetime
-from typing import Optional, List  # lint-amnesty, pylint: disable=unused-import
+from typing import Dict, FrozenSet, Optional, List, Union
 from django.contrib.auth import get_user_model
 from django.db.models.query import QuerySet
 from django.db import transaction
 from edx_django_utils.monitoring import function_trace, set_custom_attribute
 from opaque_keys import OpaqueKey
 from opaque_keys.edx.locator import LibraryLocator
-from edx_django_utils.cache import TieredCache  # lint-amnesty, pylint: disable=unused-import
-from opaque_keys.edx.keys import CourseKey  # lint-amnesty, pylint: disable=unused-import
+from edx_django_utils.cache import TieredCache
+from opaque_keys.edx.keys import CourseKey
 
 from ..data import (
     ContentErrorData,
@@ -37,6 +37,7 @@ from ..models import (
     LearningContext,
     LearningSequence,
     PublishReport,
+    UserPartitionGroup,
 )
 from .permissions import can_see_all_content
 from .processors.content_gating import ContentGatingOutlineProcessor
@@ -45,6 +46,7 @@ from .processors.schedule import ScheduleOutlineProcessor
 from .processors.special_exams import SpecialExamsOutlineProcessor
 from .processors.visibility import VisibilityOutlineProcessor
 from .processors.enrollment import EnrollmentOutlineProcessor
+from .processors.enrollment_track_partition_groups import EnrollmentTrackPartitionGroupsOutlineProcessor
 
 User = get_user_model()
 log = logging.getLogger(__name__)
@@ -111,7 +113,7 @@ def get_course_outline(course_key: CourseKey) -> CourseOutlineData:
     course_context = _get_course_context_for_outline(course_key)
 
     # Check to see if it's in the cache.
-    cache_key = "learning_sequences.api.get_course_outline.v1.{}.{}".format(
+    cache_key = "learning_sequences.api.get_course_outline.v2.{}.{}".format(
         course_context.learning_context.context_key, course_context.learning_context.published_version
     )
     outline_cache_result = TieredCache.get_cached_response(cache_key)
@@ -122,9 +124,11 @@ def get_course_outline(course_key: CourseKey) -> CourseOutlineData:
     # represented (so query CourseSection explicitly instead of relying only on
     # select_related from CourseSectionSequence).
     section_models = CourseSection.objects \
+        .prefetch_related('user_partition_groups') \
         .filter(course_context=course_context) \
         .order_by('ordering')
     section_sequence_models = CourseSectionSequence.objects \
+        .prefetch_related('user_partition_groups') \
         .filter(course_context=course_context) \
         .order_by('ordering') \
         .select_related('sequence', 'exam')
@@ -152,7 +156,10 @@ def get_course_outline(course_key: CourseKey) -> CourseOutlineData:
                 hide_from_toc=sec_seq_model.hide_from_toc,
                 visible_to_staff_only=sec_seq_model.visible_to_staff_only,
             ),
-            exam=exam_data
+            exam=exam_data,
+            user_partition_groups=_get_user_partition_groups_from_qset(
+                sec_seq_model.user_partition_groups.all()
+            ),
         )
         sec_ids_to_sequence_list[sec_seq_model.section_id].append(sequence_data)
 
@@ -164,7 +171,10 @@ def get_course_outline(course_key: CourseKey) -> CourseOutlineData:
             visibility=VisibilityData(
                 hide_from_toc=section_model.hide_from_toc,
                 visible_to_staff_only=section_model.visible_to_staff_only,
-            )
+            ),
+            user_partition_groups=_get_user_partition_groups_from_qset(
+                section_model.user_partition_groups.all()
+            ),
         )
         for section_model in section_models
     ]
@@ -185,6 +195,21 @@ def get_course_outline(course_key: CourseKey) -> CourseOutlineData:
     return outline_data
 
 
+def _get_user_partition_groups_from_qset(upg_qset) -> Dict[int, FrozenSet[int]]:
+    """
+    Given a QuerySet of UserPartitionGroup, return a mapping of UserPartition
+    IDs to the set of Group IDs for each UserPartition.
+    """
+    user_part_groups = defaultdict(set)
+    for upg in upg_qset:
+        user_part_groups[upg.partition_id].add(upg.group_id)
+
+    return {
+        partition_id: frozenset(group_ids)
+        for partition_id, group_ids in user_part_groups.items()
+    }
+
+
 def _get_course_context_for_outline(course_key: CourseKey) -> CourseContext:
     """
     Get Course Context for given param:course_key
@@ -298,6 +323,7 @@ def _get_user_course_outline_and_processors(course_key: CourseKey,  # lint-amnes
         ('special_exams', SpecialExamsOutlineProcessor),
         ('visibility', VisibilityOutlineProcessor),
         ('enrollment', EnrollmentOutlineProcessor),
+        ('enrollment_track_partitions', EnrollmentTrackPartitionGroupsOutlineProcessor),
     ]
 
     # Run each OutlineProcessor in order to figure out what items we have to
@@ -412,7 +438,7 @@ def _update_sections(course_outline: CourseOutlineData, course_context: CourseCo
     Add/Update relevant sections
     """
     for ordering, section_data in enumerate(course_outline.sections):
-        CourseSection.objects.update_or_create(
+        sec_model, _created = CourseSection.objects.update_or_create(
             course_context=course_context,
             usage_key=section_data.usage_key,
             defaults={
@@ -422,6 +448,9 @@ def _update_sections(course_outline: CourseOutlineData, course_context: CourseCo
                 'visible_to_staff_only': section_data.visibility.visible_to_staff_only,
             }
         )
+        # clear out any user partition group mappings, and remake them...
+        _update_user_partition_groups(section_data.user_partition_groups, sec_model)
+
     # Delete sections that we don't want any more
     section_usage_keys_to_keep = [
         section_data.usage_key for section_data in course_outline.sections
@@ -494,6 +523,24 @@ def _update_course_section_sequences(course_outline: CourseOutlineData, course_c
                 # Otherwise, delete any exams associated with it
                 CourseSequenceExam.objects.filter(course_section_sequence=course_section_sequence).delete()
 
+            # clear out any user partition group mappings, and remake them...
+            _update_user_partition_groups(sequence_data.user_partition_groups, course_section_sequence)
+
+
+def _update_user_partition_groups(upg_data: Dict[int, FrozenSet[int]],
+                                  model_obj: Union[CourseSection, CourseSectionSequence]):
+    """
+    Replace UserPartitionGroups associated with this content with `upg_data`.
+    """
+    model_obj.user_partition_groups.all().delete()
+    if upg_data:
+        for partition_id, group_ids in upg_data.items():
+            for group_id in group_ids:
+                upg, _ = UserPartitionGroup.objects.get_or_create(
+                    partition_id=partition_id, group_id=group_id
+                )
+                model_obj.user_partition_groups.add(upg)
+
 
 def _update_publish_report(course_outline: CourseOutlineData,
                            content_errors: List[ContentErrorData],

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

@@ -0,0 +1,83 @@
+# lint-amnesty, pylint: disable=missing-module-docstring
+import logging
+
+from common.lib.xmodule.xmodule.partitions.enrollment_track_partition_generator import (
+    create_enrollment_track_partition_with_course_id
+)
+from common.lib.xmodule.xmodule.partitions.partitions import (
+    ENROLLMENT_TRACK_PARTITION_ID,
+)
+from common.lib.xmodule.xmodule.partitions.partitions_service import get_user_partition_groups
+
+from .base import OutlineProcessor
+
+log = logging.getLogger(__name__)
+
+
+class EnrollmentTrackPartitionGroupsOutlineProcessor(OutlineProcessor):
+    """
+    Processor for applying all enrollment track user partition groups.
+
+    Confining the processor to only EnrollmentTrack user partition is a
+    significant limitation. Nonetheless, it is a step towards the goal of
+    supporting all partition schemes in the future.
+    """
+    def __init__(self, course_key, user, at_time):
+        super().__init__(course_key, user, at_time)
+        self.enrollment_track_groups = {}
+        self.user_group = None
+
+    def load_data(self):
+        """
+        Pull track groups for this course and which group the user is in.
+        """
+        user_partition = create_enrollment_track_partition_with_course_id(self.course_key)
+        self.enrollment_track_groups = get_user_partition_groups(
+            self.course_key,
+            [user_partition],
+            self.user,
+            partition_dict_key='id'
+        )
+        self.user_group = self.enrollment_track_groups.get(ENROLLMENT_TRACK_PARTITION_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?
+        """
+        if not user_partition_groups:
+            return False
+
+        groups = user_partition_groups.get(ENROLLMENT_TRACK_PARTITION_ID)
+        if not groups:
+            return False
+
+        if self.user_group and self.user_group.id not in groups:
+            # If the user's partition group, say Masters,
+            # does not belong to the partition of the block, say [verified],
+            # the block should be removed
+            return True
+        return False
+
+    def usage_keys_to_remove(self, full_course_outline):
+        """
+        Content group exclusions remove the content entirely.
+
+        If you're in the Audit track, there are things in the Verified track
+        that you don't even know exists. This processor always removes things
+        entirely instead of making them visible-but-inaccessible (like
+        ScheduleOutlineProcessor does).
+        """
+        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_data.py

@@ -125,6 +125,31 @@ class TestCourseOutlineData(TestCase):
             "A positive value will shift back the starting date for Beta users by that many days."
         )
 
+    def test_empty_user_partition_groups(self):
+        """
+        A user partition group entry with no groups is an error.
+
+        This would mean that a piece of content is associated with a partition
+        but nobody would ever be able to see it because it's not associated with
+        any group in the partition.
+        """
+        sections = generate_sections(self.course_key, [1])
+        valid_section = attr.evolve(
+            sections[0],
+            user_partition_groups={
+                50: frozenset([1, 2, 3]),
+                51: frozenset([1]),
+            }
+        )
+        with self.assertRaises(ValueError):
+            attr.evolve(
+                valid_section,
+                user_partition_groups={
+                    50: frozenset([1, 2, 3]),
+                    51: frozenset(),  # This is not allowed
+                }
+            )
+
 
 def generate_sections(course_key, num_sequences):
     """

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

@@ -6,10 +6,12 @@ from datetime import datetime, timezone
 from unittest import TestCase
 from unittest.mock import patch
 
+import ddt
 import pytest
 import attr
 from django.conf import settings
 from django.contrib.auth.models import AnonymousUser, User  # lint-amnesty, pylint: disable=imported-auth-user
+from django.db.models import signals
 from edx_proctoring.exceptions import ProctoredExamNotFoundException
 from edx_when.api import set_dates_for_course
 from mock import patch  # lint-amnesty, pylint: disable=reimported
@@ -18,9 +20,14 @@ from opaque_keys.edx.locator import LibraryLocator
 from edx_toggles.toggles.testutils import override_waffle_flag
 from openedx.core.djangolib.testing.utils import CacheIsolationTestCase
 from openedx.features.course_experience import COURSE_ENABLE_UNENROLLED_ACCESS_FLAG
+from common.djangoapps.course_modes.models import CourseMode
+from common.djangoapps.course_modes.signals import update_masters_access_course
 from common.djangoapps.student.auth import user_has_role
 from common.djangoapps.student.roles import CourseBetaTesterRole
 from common.djangoapps.student.tests.factories import BetaTesterFactory
+from common.lib.xmodule.xmodule.partitions.partitions import (
+    ENROLLMENT_TRACK_PARTITION_ID,
+)
 
 from ...data import (
     ContentErrorData,
@@ -40,6 +47,7 @@ from ..outlines import (
     key_supports_outlines,
     replace_course_outline,
 )
+from ..processors.enrollment_track_partition_groups import EnrollmentTrackPartitionGroupsOutlineProcessor
 from .test_data import generate_sections
 
 
@@ -114,9 +122,10 @@ class CourseOutlineTestCase(CacheIsolationTestCase):
         # First lets seed the data...
         replace_course_outline(self.course_outline)
 
-        # Uncached access always makes three database checks: LearningContext,
-        # CourseSection, and CourseSectionSequence.
-        with self.assertNumQueries(3):
+        # Uncached access always makes five database checks: LearningContext,
+        # CourseSection (+1 for user partition group prefetch),
+        # CourseSectionSequence (+1 for user partition group prefetch)
+        with self.assertNumQueries(5):
             uncached_outline = get_course_outline(self.course_key)
             assert uncached_outline == self.course_outline
 
@@ -137,7 +146,7 @@ class CourseOutlineTestCase(CacheIsolationTestCase):
 
         # Make sure this new outline is returned instead of the previously
         # cached one.
-        with self.assertNumQueries(3):
+        with self.assertNumQueries(5):
             uncached_new_version_outline = get_course_outline(self.course_key)  # lint-amnesty, pylint: disable=unused-variable
             assert new_version_outline == new_version_outline  # lint-amnesty, pylint: disable=comparison-with-itself
 
@@ -1280,6 +1289,405 @@ class SequentialVisibilityTestCase(CacheIsolationTestCase):
                         'Sequences should be accessible to enrolled, staff users for a public_outline course'
 
 
+@ddt.ddt
+class EnrollmentTrackPartitionGroupsTestCase(OutlineProcessorTestCase):  # lint-amnesty, pylint: disable=missing-class-docstring
+    """Tests for enrollment track partitions outline processor that affect outlines"""
+
+    @classmethod
+    def setUpTestData(cls):
+        super().setUpTestData()
+        cls.visibility = VisibilityData(
+            hide_from_toc=False,
+            visible_to_staff_only=False
+        )
+
+    def _add_course_mode(
+        self,
+        course_key,
+        mode_slug=CourseMode.VERIFIED,
+        mode_display_name='Verified Certificate'
+    ):
+        """
+        Add a course mode to the test course_key.
+        Args:
+            course_key
+            mode_slug (str): the slug of the mode to add
+            mode_display_name (str): the display name of the mode to add
+            upgrade_deadline_expired (bool): whether the upgrade deadline has passed
+        """
+        signals.post_save.disconnect(update_masters_access_course, sender=CourseMode)
+        try:
+            CourseMode.objects.create(
+                course_id=course_key,
+                mode_slug=mode_slug,
+                mode_display_name=mode_display_name,
+                min_price=50
+            )
+        finally:
+            signals.post_save.connect(update_masters_access_course, sender=CourseMode)
+
+    def _create_and_enroll_learner(self, username, mode, is_staff=False):
+        """
+        Helper function to create the learner based on the username,
+        then enroll the learner into the test course with the specified
+        mode.
+        Returns created learner
+        """
+        learner = User.objects.create_user(
+            username, email='{}@example.com'.format(username), is_staff=is_staff
+        )
+        learner.courseenrollment_set.create(course_id=self.course_key, is_active=True, mode=mode)
+        return learner
+
+    def _setup_course_outline_with_sections(
+        self,
+        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(
+            self.course_key,
+            [
+                (
+                    self.course_key.make_usage_key('course', 'course'),
+                    {'start': course_start_date}
+                )
+            ]
+        )
+
+        new_outline = CourseOutlineData(
+            course_key=self.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
+
+    def test_roundtrip(self):
+        new_outline = self._setup_course_outline_with_sections(
+            [
+                CourseSectionData(
+                    usage_key=self.course_key.make_usage_key('chapter', '0'),
+                    title="Section 0",
+                    user_partition_groups={
+                        ENROLLMENT_TRACK_PARTITION_ID: frozenset([1, 2]),
+
+                        # Just making these up to add more data
+                        51: frozenset([100]),
+                        52: frozenset([1, 2, 3, 4, 5]),
+                    }
+                )
+            ],
+        )
+
+        replace_course_outline(new_outline)
+        assert new_outline == get_course_outline(self.course_key)
+
+    @ddt.data(
+        (
+            None,
+            None,
+            [CourseMode.VERIFIED],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 1, 'student2': 1}
+        ),
+        (
+            None,
+            None,
+            [CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 1, 'student2': 1}
+        ),
+        (
+            set([2]),
+            None,
+            [CourseMode.VERIFIED],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 1, 'student2': 0}
+        ),
+        (
+            set([7]),
+            None,
+            [CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 0, 'student2': 1}
+        ),
+        (
+            set([2, 7]),
+            None,
+            [CourseMode.VERIFIED, CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 1, 'student2': 1}
+        ),
+        (
+            set([2]),
+            None,
+            [CourseMode.VERIFIED, CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 1, 'student2': 0}
+        ),
+        (
+            set([7]),
+            None,
+            [CourseMode.VERIFIED, CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 0, 'student2': 1}
+        ),
+        (
+            None,
+            set([2]),
+            [CourseMode.VERIFIED],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 1, 'student2': 0}
+        ),
+        (
+            None,
+            set([7]),
+            [CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 0, 'student2': 1}
+        ),
+        (
+            None,
+            set([2, 7]),
+            [CourseMode.VERIFIED, CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 1, 'student2': 1}
+        ),
+        (
+            None,
+            set([2]),
+            [CourseMode.VERIFIED, CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 1, 'student2': 0}
+        ),
+        (
+            None,
+            set([7]),
+            [CourseMode.VERIFIED, CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 0, 'student2': 1}
+        ),
+    )
+    @ddt.unpack
+    def test_enrollment_track_partition_on_section(
+        self,
+        section_visible_groups,
+        sequence_visible_groups,
+        course_modes,
+        learners_with_modes,
+        expected_values_dict
+    ):
+        section_user_partition_groups = None
+        sequence_user_partition_groups = None
+        if section_visible_groups:
+            section_user_partition_groups = {
+                ENROLLMENT_TRACK_PARTITION_ID: frozenset(section_visible_groups)
+            }
+        if sequence_visible_groups:
+            sequence_user_partition_groups = {
+                ENROLLMENT_TRACK_PARTITION_ID: frozenset(sequence_visible_groups)
+            }
+
+        self._setup_course_outline_with_sections(
+            [
+                CourseSectionData(
+                    usage_key=self.course_key.make_usage_key('chapter', '0'),
+                    title="Section 0",
+                    user_partition_groups=section_user_partition_groups,
+                    sequences=[
+                        CourseLearningSequenceData(
+                            usage_key=self.course_key.make_usage_key('subsection', '0'),
+                            title='Subsection 0',
+                            visibility=self.visibility,
+                            user_partition_groups=sequence_user_partition_groups,
+                        ),
+                    ]
+                )
+            ]
+        )
+
+        for course_mode in course_modes:
+            self._add_course_mode(
+                self.course_key,
+                mode_slug=course_mode,
+                mode_display_name=course_mode,
+            )
+
+        # Enroll students in the course
+        learners_to_verify = set()
+        for username, mode in learners_with_modes.items():
+            learners_to_verify.add(
+                self._create_and_enroll_learner(username, mode)
+            )
+
+        check_date = datetime(2021, 3, 27, tzinfo=timezone.utc)
+
+        # Get details
+        staff_details, _, beta_tester_details = self.get_details(check_date)
+
+        assert len(staff_details.outline.accessible_sequences) == 1
+        assert len(beta_tester_details.outline.accessible_sequences) == 0
+
+        for learner_to_verify in learners_to_verify:
+            learner_details = get_user_course_outline_details(self.course_key, learner_to_verify, check_date)
+            assert len(learner_details.outline.accessible_sequences) == expected_values_dict[learner_to_verify.username]
+
+    @ddt.data(
+        (
+            None,
+            None,
+            [CourseMode.VERIFIED],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 0, 'student2': 0}
+        ),
+        (
+            None,
+            None,
+            [CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 0, 'student2': 0}
+        ),
+        (
+            set([2]),
+            None,
+            [CourseMode.VERIFIED],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 0, 'student2': 2}
+        ),
+        (
+            set([7]),
+            None,
+            [CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 2, 'student2': 0}
+        ),
+        (
+            set([2, 7]),
+            None,
+            [CourseMode.VERIFIED, CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 0, 'student2': 0}
+        ),
+        (
+            set([2]),
+            None,
+            [CourseMode.VERIFIED, CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 0, 'student2': 2}
+        ),
+        (
+            set([7]),
+            None,
+            [CourseMode.VERIFIED, CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 2, 'student2': 0}
+        ),
+        (
+            None,
+            set([2]),
+            [CourseMode.VERIFIED],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 0, 'student2': 1}
+        ),
+        (
+            None,
+            set([7]),
+            [CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 1, 'student2': 0}
+        ),
+        (
+            None,
+            set([2, 7]),
+            [CourseMode.VERIFIED, CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 0, 'student2': 0}
+        ),
+        (
+            None,
+            set([2]),
+            [CourseMode.VERIFIED, CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 0, 'student2': 1}
+        ),
+        (
+            None,
+            set([7]),
+            [CourseMode.VERIFIED, CourseMode.MASTERS],
+            {'student1': 'verified', 'student2': 'masters'},
+            {'student1': 1, 'student2': 0}
+        ),
+    )
+    @ddt.unpack
+    def test_processor_only(
+        self,
+        section_visible_groups,
+        sequence_visible_groups,
+        course_modes,
+        learners_with_modes,
+        expected_values_dict
+    ):
+        section_user_partition_groups = None
+        sequence_user_partition_groups = None
+        if section_visible_groups:
+            section_user_partition_groups = {
+                ENROLLMENT_TRACK_PARTITION_ID: frozenset(section_visible_groups)
+            }
+        if sequence_visible_groups:
+            sequence_user_partition_groups = {
+                ENROLLMENT_TRACK_PARTITION_ID: frozenset(sequence_visible_groups)
+            }
+        full_outline = self._setup_course_outline_with_sections(
+            [
+                CourseSectionData(
+                    usage_key=self.course_key.make_usage_key('chapter', '0'),
+                    title="Section 0",
+                    user_partition_groups=section_user_partition_groups,
+                    sequences=[
+                        CourseLearningSequenceData(
+                            usage_key=self.course_key.make_usage_key('subsection', '0'),
+                            title='Subsection 0',
+                            user_partition_groups=sequence_user_partition_groups,
+                        ),
+                    ]
+                )
+            ]
+        )
+        for course_mode in course_modes:
+            self._add_course_mode(
+                self.course_key,
+                mode_slug=course_mode,
+                mode_display_name=course_mode,
+            )
+
+        # Enroll students in the course
+        learners_to_verify = set()
+        for username, mode in learners_with_modes.items():
+            learners_to_verify.add(
+                self._create_and_enroll_learner(username, mode)
+            )
+
+        check_date = datetime(2021, 3, 27, tzinfo=timezone.utc)
+        for learner_to_verify in learners_to_verify:
+            processor = EnrollmentTrackPartitionGroupsOutlineProcessor(self.course_key, learner_to_verify, check_date)
+            processor.load_data()
+            removed_usage_keys = processor.usage_keys_to_remove(full_outline)
+            assert len(removed_usage_keys) == expected_values_dict[learner_to_verify.username]
+
+
 class ContentErrorTestCase(CacheIsolationTestCase):
     """Test error collection and reporting."""
 

openedx/core/djangoapps/content/learning_sequences/data.py

@@ -21,9 +21,9 @@ Guidelines:
 TODO: Validate all datetimes to be UTC.
 """
 import logging
-from datetime import datetime  # lint-amnesty, pylint: disable=unused-import
+from datetime import datetime
 from enum import Enum
-from typing import Dict, List, Optional, Set
+from typing import Dict, FrozenSet, List, Optional
 
 import attr
 from django.contrib.auth import get_user_model
@@ -44,7 +44,6 @@ class ObjectDoesNotExist(Exception):
     Imitating Django model conventions, we put a subclass of this in some of our
     data classes to indicate when something is not found.
     """
-    pass  # lint-amnesty, pylint: disable=unnecessary-pass
 
 
 @attr.s(frozen=True, auto_attribs=True)
@@ -94,7 +93,30 @@ class ExamData:
         return self.is_practice_exam or self.is_proctored_enabled or self.is_time_limited
 
 
-@attr.s(frozen=True, auto_attribs=True)
+def user_partition_groups_not_empty(instance, attribute, value):  # pylint: disable=unused-argument
+    """
+    It's not valid to have a user_partition_groups key with no groups.
+
+    For any User Partition, users must be in one group. Associating a piece of
+    content with a user partition but no groups within that partition means that
+    the content would never be accessible to anyone who is not staff, which is
+    likely just an error. There _is_ a use case for this kind of hidden content,
+    but we do that with visible_to_staff_only.
+    """
+    # If it's None or an empty dictionary, we don't have to check it.
+    if not value:
+        return
+
+    for partition_id, group_list in value.items():
+        if not group_list:
+            raise ValueError(
+                f"{instance.usage_key} has an empty list of groups for "
+                f"user_partition_groups[{partition_id}]. User Partitioned "
+                f"content must be associated with at least one group."
+            )
+
+
+@attr.s(frozen=True)
 class CourseLearningSequenceData:
     """
     A Learning Sequence (a.k.a. subsection) from a Course.
@@ -104,22 +126,40 @@ class CourseLearningSequenceData:
     learning sequences in Courses vs. Pathways vs. Libraries. Such an object
     would likely not have `visibility` as that holds course-specific concepts.
     """
-    usage_key: UsageKey
-    title: str
-    visibility: VisibilityData = VisibilityData()
-    exam: ExamData = ExamData()
-    inaccessible_after_due: bool = False
+    usage_key = attr.ib(type=UsageKey)
+    title = attr.ib(type=str)
+    visibility = attr.ib(type=VisibilityData, default=VisibilityData())
+    exam = attr.ib(type=ExamData, default=ExamData())
+    inaccessible_after_due = attr.ib(type=bool, default=False)
+
+    # Mapping of UserPartition IDs to list of UserPartition Groups that are
+    # associated with this piece of content. See models.UserPartitionGroup
+    # for more details.
+    user_partition_groups = attr.ib(
+        type=Dict[int, FrozenSet[int]],
+        factory=dict,
+        validator=[user_partition_groups_not_empty],
+    )
 
 
-@attr.s(frozen=True, auto_attribs=True)
+@attr.s(frozen=True)
 class CourseSectionData:
     """
     A Section in a Course (sometimes called a Chapter).
     """
-    usage_key: UsageKey
-    title: str
-    visibility: VisibilityData
-    sequences: List[CourseLearningSequenceData]
+    usage_key = attr.ib(type=UsageKey)
+    title = attr.ib(type=str)
+    visibility = attr.ib(type=VisibilityData, default=VisibilityData())
+    sequences = attr.ib(type=List[CourseLearningSequenceData], factory=list)
+
+    # Mapping of UserPartition IDs to list of UserPartition Groups that are
+    # associated with this piece of content. See models.UserPartitionGroup
+    # for more details.
+    user_partition_groups = attr.ib(
+        type=Dict[int, FrozenSet[int]],
+        factory=dict,
+        validator=[user_partition_groups_not_empty],
+    )
 
 
 @attr.s(frozen=True)
@@ -314,7 +354,7 @@ class UserCourseOutlineData(CourseOutlineData):
     # * If anonymous course access is enabled in "public_outline" mode,
     #   unauthenticated users (AnonymousUser) will see the course outline but
     #   not be able to access anything inside.
-    accessible_sequences: Set[UsageKey]
+    accessible_sequences: FrozenSet[UsageKey]
 
 
 @attr.s(frozen=True, auto_attribs=True)

openedx/core/djangoapps/content/learning_sequences/migrations/0012_add_user_partition_group.py

@@ -0,0 +1,35 @@
+# Generated by Django 2.2.19 on 2021-03-26 13:50
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('learning_sequences', '0011_course_meta_names'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='UserPartitionGroup',
+            fields=[
+                ('id', models.BigAutoField(primary_key=True, serialize=False)),
+                ('partition_id', models.BigIntegerField()),
+                ('group_id', models.BigIntegerField()),
+            ],
+        ),
+        migrations.AddIndex(
+            model_name='userpartitiongroup',
+            index=models.Index(fields=['partition_id', 'group_id'], name='learning_se_partiti_6e2d28_idx'),
+        ),
+        migrations.AddField(
+            model_name='coursesection',
+            name='user_partition_groups',
+            field=models.ManyToManyField(to='learning_sequences.UserPartitionGroup'),
+        ),
+        migrations.AddField(
+            model_name='coursesectionsequence',
+            name='user_partition_groups',
+            field=models.ManyToManyField(to='learning_sequences.UserPartitionGroup'),
+        ),
+    ]

openedx/core/djangoapps/content/learning_sequences/models.py

@@ -148,12 +148,46 @@ class CourseContentVisibilityMixin(models.Model):
         abstract = True
 
 
+class UserPartitionGroup(models.Model):
+    """
+    Each row represents a Group in a UserPartition.
+
+    UserPartitions is a pluggable interface. Some IDs are static (with values
+    less than 100). Others are dynamic, picking a range between 100 and 2^31-1.
+    That means that technically, we could use IntegerField instead of
+    BigIntegerField, but a) that limit isn't actually enforced as far as I can
+    tell; and b) it's not _that_ much extra storage, so I'm using BigInteger
+    instead (2^63-1).
+
+    It's a pluggable interface (entry points: openedx.user_partition_scheme,
+    openedx.dynamic_partition_generator), so there's no "UserPartition" model.
+    We need to actually link this against the values passed back from the
+    partitions service in order to map them to names and descriptions.
+
+    Any CourseSection or CourseSectionSequence may be associated with any number
+    of UserPartitionGroups. An individual _user_ may only be in one Group for
+    any given User Partition, but a piece of _content_ can be associated with
+    multiple groups. So for instance, for the Enrollment Track user partition,
+    a piece of content may be associated with both "Verified" and "Masters"
+    tracks, while a user may only be in one or the other.
+
+    UserPartitionGroups are not associated with LearningSequence directly
+    because User Partitions often carry course-level assumptions (e.g.
+    Enrollment Track) that don't make sense outside of a Course.
+    """
+    id = models.BigAutoField(primary_key=True)
+    partition_id = models.BigIntegerField(null=False)
+    group_id = models.BigIntegerField(null=False)
+
+    class Meta:
+        indexes = [
+            models.Index(fields=['partition_id', 'group_id']),
+        ]
+
+
 class CourseSection(CourseContentVisibilityMixin, TimeStampedModel):
     """
     Course Section data, mapping to the 'chapter' block type.
-
-    Do NOT make a foreign key against this table, as the values are deleted and
-    re-created on course publish.
     """
     id = models.BigAutoField(primary_key=True)
     course_context = models.ForeignKey(
@@ -165,6 +199,8 @@ class CourseSection(CourseContentVisibilityMixin, TimeStampedModel):
     # What is our position within the Course? (starts with 0)
     ordering = models.PositiveIntegerField(null=False)
 
+    user_partition_groups = models.ManyToManyField(UserPartitionGroup)
+
     class Meta:
         unique_together = [
             ['course_context', 'usage_key'],
@@ -203,6 +239,8 @@ class CourseSectionSequence(CourseContentVisibilityMixin, TimeStampedModel):
     # sequences across 20 sections, the numbering here would be 0-199.
     ordering = models.PositiveIntegerField(null=False)
 
+    user_partition_groups = models.ManyToManyField(UserPartitionGroup)
+
     class Meta:
         unique_together = [
             ['course_context', 'ordering'],