diff --git a/.gitignore b/.gitignore index 46d8147..6a6fd92 100644 --- a/.gitignore +++ b/.gitignore @@ -63,3 +63,6 @@ docs/openedx_user_groups.*.rst # Private requirements requirements/private.in requirements/private.txt + +# VSCode +.vscode/ diff --git a/openedx_user_groups/partitions/__init__.py b/openedx_user_groups/partitions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/openedx_user_groups/partitions/user_group_partition_scheme.py b/openedx_user_groups/partitions/user_group_partition_scheme.py new file mode 100644 index 0000000..fb87782 --- /dev/null +++ b/openedx_user_groups/partitions/user_group_partition_scheme.py @@ -0,0 +1,213 @@ +""" +Provides a UserPartition driver for user groups. +""" + +import logging +from unittest.mock import Mock + +from django.utils.translation import gettext_lazy as _ + +try: + from lms.djangoapps.courseware.masquerade import ( + get_course_masquerade, + get_masquerading_user_group, + is_masquerading_as_specific_student, + ) + from openedx.core import types + from xmodule.partitions.partitions import Group, UserPartition, UserPartitionError +except ImportError: + get_course_masquerade = Mock() + get_masquerading_user_group = Mock() + is_masquerading_as_specific_student = Mock() + types = Mock() + Group = Mock() + + class UserPartitionError(Exception): + """Mock UserPartitionError class for testing.""" + + class UserPartition: + """Mock UserPartition class for testing.""" + + # pylint: disable=redefined-builtin, too-many-positional-arguments, unused-argument + def __init__(self, id, name, description, groups, scheme, parameters, active=True): + self.parameters = parameters + self.scheme = scheme + self.id = id + self.name = name + self.description = description + self.active = active + + @property + def groups(self): + return self.groups + + +from opaque_keys.edx.keys import CourseKey + +from openedx_user_groups.models import UserGroup, UserGroupMembership +from openedx_user_groups.toggles import is_user_groups_enabled + +log = logging.getLogger(__name__) + +# TODO: This is a temporary ID. We should use a more permanent ID. +USER_GROUP_PARTITION_ID = 1000000000 +USER_GROUP_SCHEME = "user_group" + + +class UserGroupPartition(UserPartition): + """ + Extends UserPartition to support dynamic groups pulled from the new user + groups system. + """ + + @property + def groups(self) -> list[Group]: + """ + Dynamically generate groups (based on user groups) for this partition. + """ + course_key = CourseKey.from_string(self.parameters["course_id"]) + if not is_user_groups_enabled(course_key): + return [] + + # TODO: Only get user groups for the course. + user_groups = UserGroup.objects.filter(enabled=True) + return [Group(user_group.id, str(user_group.name)) for user_group in user_groups] + + +class UserGroupPartitionScheme: + """Uses user groups to map learners into partition groups. + + This scheme is only available if the ENABLE_USER_GROUPS waffle flag is enabled for the course. + + This is how it works: + - A only one user partition is created for each course with the `USER_GROUP_PARTITION_ID`. + - A (Content) group is created for each user group in the course with the + database user group ID as the group ID, and the user group name as the + group name. + - A user is assigned to a group if they are a member of the user group. + """ + + # TODO: A user could belong to multiple groups. This method assumes that + # the user belongs to a single group. This should be renamed? + @classmethod + def get_group_for_user( + cls, course_key: CourseKey, user: types.User, user_partition: UserPartition + ) -> list[Group] | None: + """Get the (User) Group from the specified user partition for the user. + + A user is assigned to the group via their user group membership and any + mappings from user groups to partitions / groups that might exist. + + Args: + course_key (CourseKey): The course key. + user (types.User): The user. + user_partition (UserPartition): The user partition. + + Returns: + List[Group]: The groups in the specified user partition for the user. + None if the user is not a member of any group. + """ + if not is_user_groups_enabled(course_key): + return None + + 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) + + user_group_ids = UserGroupMembership.objects.filter(user=user, is_active=True).values_list( + "group__id", flat=True + ) + all_user_groups: list[UserGroup] = UserGroup.objects.filter(enabled=True) + + if not user_group_ids: + return None + + user_groups = [] + for user_group in all_user_groups: + if user_group.id in user_group_ids: + user_groups.append(Group(user_group.id, str(user_group.name))) + + return user_groups + + # pylint: disable=redefined-builtin, invalid-name, too-many-positional-arguments + @classmethod + def create_user_partition( + cls, + id: int, + name: str, + description: str, + groups: list[Group] | None = None, + parameters: dict | None = None, + active: bool = True, + ) -> UserPartition: + """Create a custom UserPartition to support dynamic groups based on user groups. + + 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: + UserGroupPartition: The user partition. + """ + course_key = CourseKey.from_string(parameters["course_id"]) + if not is_user_groups_enabled(course_key): + return None + + user_group_partition = UserGroupPartition( + id, + str(name), + str(description), + groups, + cls, + parameters, + active=active, + ) + + return user_group_partition + + +def create_user_group_partition_with_course_id(course_id: CourseKey) -> UserPartition | None: + """ + Create and return the user group partition based only on course_id. + If it cannot be created, None is returned. + """ + try: + user_group_scheme = UserPartition.get_scheme(USER_GROUP_SCHEME) + except UserPartitionError: + log.warning(f"No {USER_GROUP_SCHEME} scheme registered, UserGroupPartition will not be created.") + return None + + partition = user_group_scheme.create_user_partition( + id=USER_GROUP_PARTITION_ID, + name=_("User Groups"), + description=_("Partition for segmenting users by user groups"), + parameters={"course_id": str(course_id)}, + ) + + return partition + + +def create_user_group_partition(course): + """ + Get the dynamic user group user partition based on the user groups of the course. + """ + if not is_user_groups_enabled(course.id): + return [] + + return create_user_group_partition_with_course_id(course.id) diff --git a/openedx_user_groups/processors/__init__.py b/openedx_user_groups/processors/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/openedx_user_groups/processors/user_group_partition_groups.py b/openedx_user_groups/processors/user_group_partition_groups.py new file mode 100644 index 0000000..98d6c22 --- /dev/null +++ b/openedx_user_groups/processors/user_group_partition_groups.py @@ -0,0 +1,123 @@ +""" +Outline processors for applying user group partition groups. +""" + +from datetime import datetime +from typing import Dict, List, Set +from unittest.mock import Mock + +from opaque_keys.edx.keys import CourseKey + +try: + from openedx.core import types + from openedx.core.djangoapps.content.learning_sequences.api.processors.base import OutlineProcessor + from xmodule.partitions.partitions import Group + from xmodule.partitions.partitions_service import get_user_partition_groups +except ImportError: + types = Mock() + Group = Mock() + get_user_partition_groups = Mock() + + class OutlineProcessor: + """Mock OutlineProcessor class.""" + + def __init__(self, course_key, user, at_time): + """Initialize the OutlineProcessor.""" + self.course_key = course_key + self.user = user + self.at_time = at_time + + +from openedx_user_groups.partitions.user_group_partition_scheme import ( + USER_GROUP_PARTITION_ID, + create_user_group_partition_with_course_id, +) +from openedx_user_groups.toggles import is_user_groups_enabled + + +class UserGroupPartitionGroupsOutlineProcessor(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 user group 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): + """ + Initialize the UserGroupPartitionGroupsOutlineProcessor. + + Args: + course_key (CourseKey): The course key. + user (types.User): The user. + at_time (datetime): The time at which the data is loaded. + """ + super().__init__(course_key, user, at_time) + self.user_groups: List[Group] = [] + + def load_data(self, _) -> None: + """ + Pull user groups for this course and which group the user is in. + """ + if not is_user_groups_enabled(self.course_key): + return + + user_partition = create_user_group_partition_with_course_id(self.course_key) + self.user_groups = get_user_partition_groups( + self.course_key, + [user_partition], + self.user, + partition_dict_key="id", + ).get(USER_GROUP_PARTITION_ID) + + def _is_user_excluded_by_partition_group(self, user_partition_groups: Dict[int, Set[int]]): + """ + Is the user part of the group to which the block is restricting content? + + Args: + 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 is_user_groups_enabled(self.course_key): + return False + + if not user_partition_groups: + return False + + groups = user_partition_groups.get(USER_GROUP_PARTITION_ID) + if not groups: + return False + + for group in self.user_groups: + if group.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 user group membership. + """ + 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 diff --git a/openedx_user_groups/toggles.py b/openedx_user_groups/toggles.py new file mode 100644 index 0000000..9ebf326 --- /dev/null +++ b/openedx_user_groups/toggles.py @@ -0,0 +1,39 @@ +""" +Toggles for user groups. +This module defines feature flags (waffle flags) used to enable or disable functionality related to user groups +within the Open edX platform. These toggles allow for dynamic control of features without requiring code changes. +""" + +from opaque_keys.edx.keys import CourseKey + +try: + from openedx.core.djangoapps.waffle_utils import CourseWaffleFlag +except ImportError: + + class CourseWaffleFlag: + """Mock CourseWaffleFlag class.""" + + def __init__(self, name, module_name): + """Initialize the CourseWaffleFlag.""" + self.name = name + self.module_name = module_name + + +# Namespace for all user group related waffle flags +WAFFLE_FLAG_NAMESPACE = "user_groups" + +# .. toggle_name: user_groups.enable_user_groups +# .. toggle_implementation: CourseWaffleFlag +# .. toggle_default: False +# .. toggle_description: Waffle flag to enable or disable the user groups feature in a course. +# .. toggle_use_cases: temporary, open_edx +# .. toggle_creation_date: 2025-06-19 +# .. toggle_target_removal_date: None +ENABLE_USER_GROUPS = CourseWaffleFlag(f"{WAFFLE_FLAG_NAMESPACE}.enable_user_groups", __name__) + + +def is_user_groups_enabled(course_key: CourseKey) -> bool: + """ + Returns a boolean if user groups are enabled for the course. + """ + return ENABLE_USER_GROUPS.is_enabled(course_key) diff --git a/requirements/base.in b/requirements/base.in index 4ff44c7..c75ce1f 100644 --- a/requirements/base.in +++ b/requirements/base.in @@ -3,10 +3,10 @@ Django # Web application framework edx-django-utils # edX utilities for Django - openedx-atlas openedx-events # Open edX Events library for updating user groups celery # Celery for background tasks djangorestframework edx-organizations pydantic +edx-opaque-keys # Open edX opaque keys library diff --git a/requirements/base.txt b/requirements/base.txt index 4b5f252..0d8964e 100644 --- a/requirements/base.txt +++ b/requirements/base.txt @@ -1,5 +1,5 @@ # -# This file is autogenerated by pip-compile with Python 3.12 +# This file is autogenerated by pip-compile with Python 3.11 # by the following command: # # pip-compile --output-file=requirements/base.txt requirements/base.in @@ -85,6 +85,7 @@ edx-drf-extensions==10.6.0 # via edx-organizations edx-opaque-keys[django]==3.0.0 # via + # -r requirements/base.in # edx-ccx-keys # edx-drf-extensions # edx-organizations diff --git a/requirements/ci.txt b/requirements/ci.txt index 2802d13..5338cc3 100644 --- a/requirements/ci.txt +++ b/requirements/ci.txt @@ -1,5 +1,5 @@ # -# This file is autogenerated by pip-compile with Python 3.12 +# This file is autogenerated by pip-compile with Python 3.11 # by the following command: # # pip-compile --output-file=requirements/ci.txt requirements/ci.in diff --git a/requirements/dev.txt b/requirements/dev.txt index c4b79c2..2fd87d3 100644 --- a/requirements/dev.txt +++ b/requirements/dev.txt @@ -1,5 +1,5 @@ # -# This file is autogenerated by pip-compile with Python 3.12 +# This file is autogenerated by pip-compile with Python 3.11 # by the following command: # # pip-compile --output-file=requirements/dev.txt requirements/dev.in diff --git a/requirements/doc.txt b/requirements/doc.txt index 5a37b8a..34d60f6 100644 --- a/requirements/doc.txt +++ b/requirements/doc.txt @@ -1,5 +1,5 @@ # -# This file is autogenerated by pip-compile with Python 3.12 +# This file is autogenerated by pip-compile with Python 3.11 # by the following command: # # pip-compile --output-file=requirements/doc.txt requirements/doc.in @@ -28,6 +28,8 @@ babel==2.17.0 # via # pydata-sphinx-theme # sphinx +backports-tarfile==1.2.0 + # via jaraco-context beautifulsoup4==4.13.4 # via pydata-sphinx-theme billiard==4.2.1 @@ -179,6 +181,8 @@ idna==3.10 # requests imagesize==1.4.1 # via sphinx +importlib-metadata==8.7.0 + # via keyring iniconfig==2.1.0 # via # -r requirements/test.txt @@ -412,6 +416,8 @@ wcwidth==0.2.13 # via # -r requirements/test.txt # prompt-toolkit +zipp==3.23.0 + # via importlib-metadata # The following packages are considered to be unsafe in a requirements file: # setuptools diff --git a/requirements/pip-tools.txt b/requirements/pip-tools.txt index f382002..ff2aa7b 100644 --- a/requirements/pip-tools.txt +++ b/requirements/pip-tools.txt @@ -1,5 +1,5 @@ # -# This file is autogenerated by pip-compile with Python 3.12 +# This file is autogenerated by pip-compile with Python 3.11 # by the following command: # # pip-compile --output-file=requirements/pip-tools.txt requirements/pip-tools.in diff --git a/requirements/pip.txt b/requirements/pip.txt index e621829..59d8969 100644 --- a/requirements/pip.txt +++ b/requirements/pip.txt @@ -1,5 +1,5 @@ # -# This file is autogenerated by pip-compile with Python 3.12 +# This file is autogenerated by pip-compile with Python 3.11 # by the following command: # # pip-compile --allow-unsafe --output-file=requirements/pip.txt requirements/pip.in diff --git a/requirements/quality.txt b/requirements/quality.txt index 6a5097c..4b445bc 100644 --- a/requirements/quality.txt +++ b/requirements/quality.txt @@ -1,5 +1,5 @@ # -# This file is autogenerated by pip-compile with Python 3.12 +# This file is autogenerated by pip-compile with Python 3.11 # by the following command: # # pip-compile --output-file=requirements/quality.txt requirements/quality.in diff --git a/requirements/test.txt b/requirements/test.txt index 3c90eca..83b5e61 100644 --- a/requirements/test.txt +++ b/requirements/test.txt @@ -1,5 +1,5 @@ # -# This file is autogenerated by pip-compile with Python 3.12 +# This file is autogenerated by pip-compile with Python 3.11 # by the following command: # # pip-compile --output-file=requirements/test.txt requirements/test.in diff --git a/setup.py b/setup.py index 0ca303b..fbbf392 100755 --- a/setup.py +++ b/setup.py @@ -180,5 +180,11 @@ def is_requirement(line): "lms.djangoapp": [ "openedx_user_groups = openedx_user_groups.apps:OpenedxUserGroupsConfig", ], + "openedx.user_partition_scheme": [ + "user_group = openedx_user_groups.partitions.user_group_partition_scheme:UserGroupPartitionScheme", + ], + "openedx.dynamic_partition_generator": [ + "user_group = openedx_user_groups.partitions.user_group_partition_scheme:create_user_group_partition", + ], }, ) diff --git a/tests/test_toggles.py b/tests/test_toggles.py new file mode 100644 index 0000000..37782ec --- /dev/null +++ b/tests/test_toggles.py @@ -0,0 +1,165 @@ +""" +Test Suite for the User Group toggles. + +This test suite covers all toggle functionality defined in toggles.py. +""" + +from unittest.mock import Mock, patch + +from django.test import TestCase +from opaque_keys.edx.keys import CourseKey + +from openedx_user_groups.toggles import ENABLE_USER_GROUPS, WAFFLE_FLAG_NAMESPACE, is_user_groups_enabled + + +class TestToggles(TestCase): + """Test toggle functionality and configuration.""" + + def test_waffle_flag_namespace(self): + """Test that the waffle flag namespace is correctly defined. + + Expected Results: + - The namespace should be "user_groups". + """ + self.assertEqual(WAFFLE_FLAG_NAMESPACE, "user_groups") + + def test_enable_user_groups_flag_creation(self): + """Test that the ENABLE_USER_GROUPS flag is created with correct parameters. + + Expected Results: + - The flag should be created with the correct namespace and module name. + """ + expected_flag_name = f"{WAFFLE_FLAG_NAMESPACE}.enable_user_groups" + self.assertEqual(ENABLE_USER_GROUPS.name, expected_flag_name) + self.assertEqual(ENABLE_USER_GROUPS.module_name, "openedx_user_groups.toggles") + + @patch("openedx_user_groups.toggles.ENABLE_USER_GROUPS") + def test_is_user_groups_enabled_when_flag_enabled(self, mock_flag: Mock): + """Test is_user_groups_enabled returns True when flag is enabled. + + Expected Results: + - The function should return True when the waffle flag is enabled. + """ + course_key = CourseKey.from_string("course-v1:edX+Demo+Course") + mock_flag.is_enabled.return_value = True + + result = is_user_groups_enabled(course_key) + + self.assertTrue(result) + mock_flag.is_enabled.assert_called_once_with(course_key) + + @patch("openedx_user_groups.toggles.ENABLE_USER_GROUPS") + def test_is_user_groups_enabled_when_flag_disabled(self, mock_flag: Mock): + """Test is_user_groups_enabled returns False when flag is disabled. + + Expected Results: + - The function should return False when the waffle flag is disabled. + """ + course_key = CourseKey.from_string("course-v1:edX+Demo+Course") + mock_flag.is_enabled.return_value = False + + result = is_user_groups_enabled(course_key) + + self.assertFalse(result) + mock_flag.is_enabled.assert_called_once_with(course_key) + + @patch("openedx_user_groups.toggles.ENABLE_USER_GROUPS") + def test_is_user_groups_enabled_with_different_course_keys(self, mock_flag: Mock): + """Test is_user_groups_enabled works with different course key formats. + + Expected Results: + - The function should work correctly with different course key formats. + """ + course_keys = [ + CourseKey.from_string("course-v1:edX+Demo+Course"), + CourseKey.from_string("course-v1:TestOrg+CS101+2024"), + CourseKey.from_string("course-v1:AnotherOrg+Math101+Fall2024"), + ] + mock_flag.is_enabled.return_value = True + + for course_key in course_keys: + result = is_user_groups_enabled(course_key) + self.assertTrue(result) + mock_flag.is_enabled.assert_called_with(course_key) + + @patch("openedx_user_groups.toggles.ENABLE_USER_GROUPS") + def test_is_user_groups_enabled_flag_called_with_correct_arguments(self, mock_flag: Mock): + """Test that the flag is called with the correct course key argument. + + Expected Results: + - The waffle flag should be called with the exact course key passed to the function. + """ + course_key = CourseKey.from_string("course-v1:edX+Demo+Course") + mock_flag.is_enabled.return_value = True + + is_user_groups_enabled(course_key) + + mock_flag.is_enabled.assert_called_once_with(course_key) + # Verify the course key is the same object, not just equal + called_course_key = mock_flag.is_enabled.call_args[0][0] + self.assertEqual(called_course_key, course_key) + + +class TestToggleIntegration(TestCase): + """Test toggle integration with course keys and real scenarios.""" + + def setUp(self): + """Set up test data.""" + self.course_key = CourseKey.from_string("course-v1:edX+Demo+Course") + + @patch("openedx_user_groups.toggles.ENABLE_USER_GROUPS") + def test_toggle_with_real_course_key_objects(self, mock_flag: Mock): + """Test toggle functionality with real CourseKey objects. + + Expected Results: + - The toggle should work correctly with real CourseKey objects. + """ + mock_flag.is_enabled.return_value = True + + result = is_user_groups_enabled(self.course_key) + + self.assertTrue(result) + mock_flag.is_enabled.assert_called_once_with(self.course_key) + + @patch("openedx_user_groups.toggles.ENABLE_USER_GROUPS") + def test_toggle_behavior_consistency(self, mock_flag: Mock): + """Test that toggle behavior is consistent across multiple calls. + + Expected Results: + - Multiple calls with the same course key should return the same result. + """ + mock_flag.is_enabled.return_value = True + + result1 = is_user_groups_enabled(self.course_key) + result2 = is_user_groups_enabled(self.course_key) + result3 = is_user_groups_enabled(self.course_key) + + self.assertTrue(result1) + self.assertTrue(result2) + self.assertTrue(result3) + self.assertEqual(mock_flag.is_enabled.call_count, 3) + + @patch("openedx_user_groups.toggles.ENABLE_USER_GROUPS") + def test_toggle_with_different_course_keys_returns_different_results(self, mock_flag: Mock): + """Test that different course keys can have different toggle states. + + Expected Results: + - Different course keys can have different toggle states. + """ + course_key1 = CourseKey.from_string("course-v1:edX+Demo+Course") + course_key2 = CourseKey.from_string("course-v1:TestOrg+CS101+2024") + + # Configure mock to return different values for different course keys + def mock_is_enabled(course_key): + if str(course_key) == "course-v1:edX+Demo+Course": + return True + return False + + mock_flag.is_enabled.side_effect = mock_is_enabled + + result1 = is_user_groups_enabled(course_key1) + result2 = is_user_groups_enabled(course_key2) + + self.assertTrue(result1) + self.assertFalse(result2) + self.assertEqual(mock_flag.is_enabled.call_count, 2) diff --git a/tests/test_user_group_partition_groups.py b/tests/test_user_group_partition_groups.py new file mode 100644 index 0000000..acde75c --- /dev/null +++ b/tests/test_user_group_partition_groups.py @@ -0,0 +1,335 @@ +""" +Unit tests for UserGroupPartitionGroupsOutlineProcessor. + +This test suite covers all methods and behaviors of the UserGroupPartitionGroupsOutlineProcessor +class, including initialization, data loading, partition group exclusion logic, and usage key removal. +""" + +from datetime import datetime +from unittest.mock import Mock, patch + +from django.test import TestCase +from opaque_keys.edx.keys import CourseKey + +from openedx_user_groups.processors.user_group_partition_groups import ( + USER_GROUP_PARTITION_ID, + UserGroupPartitionGroupsOutlineProcessor, +) +from tests.factories import UserFactory + + +class GroupMock: + """Mock class for course outline sections.""" + + def __init__(self, id, name): # pylint: disable=redefined-builtin + self.id = id + self.name = name + + def __eq__(self, other): + """Compare GroupMock instances by their attributes.""" + if not isinstance(other, GroupMock): + return False + return self.id == other.id and self.name == other.name + + +class MockOutlineSection: + """Mock class for course outline sections.""" + + def __init__(self, usage_key, user_partition_groups=None): + self.usage_key = usage_key + self.user_partition_groups = user_partition_groups or {} + self.sequences = [] + + +class MockOutlineSequence: + """Mock class for course outline sequences.""" + + def __init__(self, usage_key, user_partition_groups=None): + self.usage_key = usage_key + self.user_partition_groups = user_partition_groups or {} + + +class MockFullCourseOutline: + """Mock class for full course outline.""" + + def __init__(self, sections=None): + self.sections = sections or [] + + +# pylint: disable=protected-access +class TestUserGroupPartitionGroupsOutlineProcessor(TestCase): + """Test suite for UserGroupPartitionGroupsOutlineProcessor.""" + + def setUp(self): + """Set up test data.""" + self.course_key = CourseKey.from_string("course-v1:edX+Demo+Course") + self.user = UserFactory() + self.at_time = datetime.now() + self.user_group_1 = GroupMock(id=1, name="Test Group 1") + self.user_group_2 = GroupMock(id=2, name="Test Group 2") + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_initialization(self, mock_is_enabled: Mock): + """Test processor initialization.""" + mock_is_enabled.return_value = True + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + + self.assertEqual(processor.course_key, self.course_key) + self.assertEqual(processor.user, self.user) + self.assertEqual(processor.at_time, self.at_time) + self.assertEqual(processor.user_groups, []) + + @patch("openedx_user_groups.processors.user_group_partition_groups.create_user_group_partition_with_course_id") + @patch("openedx_user_groups.processors.user_group_partition_groups.get_user_partition_groups") + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_load_data_when_user_groups_enabled( + self, mock_is_enabled: Mock, mock_get_groups: Mock, mock_create_partition: Mock + ): + """Test load_data method when user groups are enabled.""" + mock_is_enabled.return_value = True + # Mock partition and groups + mock_partition = Mock() + mock_create_partition.return_value = mock_partition + mock_group_1 = GroupMock(id=1, name="Test Group 1") + mock_group_2 = GroupMock(id=2, name="Test Group 2") + mock_get_groups.return_value = {USER_GROUP_PARTITION_ID: {1: mock_group_1, 2: mock_group_2}} + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + processor.load_data(None) + + mock_create_partition.assert_called_once_with(self.course_key) + mock_get_groups.assert_called_once_with(self.course_key, [mock_partition], self.user, partition_dict_key="id") + self.assertEqual(len(processor.user_groups), 2) + self.assertIn(1, processor.user_groups) + self.assertIn(2, processor.user_groups) + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_load_data_when_user_groups_disabled(self, mock_is_enabled: Mock): + """Test load_data method when user groups are disabled.""" + mock_is_enabled.return_value = False + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + processor.load_data(None) + + self.assertEqual(processor.user_groups, []) + + @patch("openedx_user_groups.processors.user_group_partition_groups.create_user_group_partition_with_course_id") + @patch("openedx_user_groups.processors.user_group_partition_groups.get_user_partition_groups") + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_load_data_when_no_user_groups_found( + self, mock_is_enabled: Mock, mock_get_groups: Mock, mock_create_partition: Mock + ): + """Test load_data method when no user groups are found.""" + mock_is_enabled.return_value = True + mock_partition = Mock() + mock_create_partition.return_value = mock_partition + mock_get_groups.return_value = {} + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + processor.load_data(None) + + self.assertIsNone(processor.user_groups) + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_is_user_excluded_by_partition_group_when_disabled(self, mock_is_enabled: Mock): + """Test _is_user_excluded_by_partition_group when user groups are disabled.""" + mock_is_enabled.return_value = False + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + result = processor._is_user_excluded_by_partition_group({1: {1, 2}}) + + self.assertFalse(result) + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_is_user_excluded_by_partition_group_when_no_partition_groups(self, mock_is_enabled: Mock): + """Test _is_user_excluded_by_partition_group when no partition groups are provided.""" + mock_is_enabled.return_value = True + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + result = processor._is_user_excluded_by_partition_group({}) + + self.assertFalse(result) + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_is_user_excluded_by_partition_group_when_no_user_group_partition(self, mock_is_enabled: Mock): + """Test _is_user_excluded_by_partition_group when no user group partition is found.""" + mock_is_enabled.return_value = True + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + result = processor._is_user_excluded_by_partition_group({999: {1, 2}}) + + self.assertFalse(result) + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_is_user_excluded_by_partition_group_when_user_in_group(self, mock_is_enabled: Mock): + """Test _is_user_excluded_by_partition_group when user is in one of the groups.""" + mock_is_enabled.return_value = True + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + processor.user_groups = [self.user_group_1, self.user_group_2] + result = processor._is_user_excluded_by_partition_group({USER_GROUP_PARTITION_ID: {1, 3}}) + + self.assertFalse(result) + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_is_user_excluded_by_partition_group_when_user_not_in_group(self, mock_is_enabled: Mock): + """Test _is_user_excluded_by_partition_group when user is not in any of the groups.""" + mock_is_enabled.return_value = True + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + processor.user_groups = [self.user_group_1, self.user_group_2] + result = processor._is_user_excluded_by_partition_group({USER_GROUP_PARTITION_ID: {3, 4}}) + + self.assertTrue(result) + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_usage_keys_to_remove_with_no_exclusions(self, mock_is_enabled: Mock): + """Test usage_keys_to_remove when no content should be excluded.""" + mock_is_enabled.return_value = True + # Create mock outline with sections and sequences + section1 = MockOutlineSection("section1", {}) + section1.sequences = [MockOutlineSequence("seq1", {}), MockOutlineSequence("seq2", {})] + section2 = MockOutlineSection("section2", {}) + section2.sequences = [MockOutlineSequence("seq3", {})] + full_course_outline = MockFullCourseOutline([section1, section2]) + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + result = processor.usage_keys_to_remove(full_course_outline) + + self.assertEqual(result, set()) + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_usage_keys_to_remove_with_section_exclusion(self, mock_is_enabled: Mock): + """Test usage_keys_to_remove when a section should be excluded.""" + mock_is_enabled.return_value = True + # Create mock outline with one excluded section + section1 = MockOutlineSection( + "section1", + {USER_GROUP_PARTITION_ID: {3, 4}}, # User not in these groups + ) + section1.sequences = [MockOutlineSequence("seq1", {}), MockOutlineSequence("seq2", {})] + section2 = MockOutlineSection("section2", {}) + section2.sequences = [MockOutlineSequence("seq3", {})] + full_course_outline = MockFullCourseOutline([section1, section2]) + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + processor.user_groups = [self.user_group_1, self.user_group_2] + result = processor.usage_keys_to_remove(full_course_outline) + + expected = {"section1", "seq1", "seq2"} + self.assertEqual(result, expected) + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_usage_keys_to_remove_with_sequence_exclusion(self, mock_is_enabled: Mock): + """Test usage_keys_to_remove when a sequence should be excluded.""" + mock_is_enabled.return_value = True + # Create mock outline with one excluded sequence + section1 = MockOutlineSection("section1", {}) + section1.sequences = [ + MockOutlineSequence( + "seq1", + {USER_GROUP_PARTITION_ID: {3, 4}}, # User not in these groups + ), + MockOutlineSequence("seq2", {}), + ] + section2 = MockOutlineSection("section2", {}) + section2.sequences = [MockOutlineSequence("seq3", {})] + full_course_outline = MockFullCourseOutline([section1, section2]) + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + processor.user_groups = [self.user_group_1, self.user_group_2] + result = processor.usage_keys_to_remove(full_course_outline) + + expected = {"seq1"} + self.assertEqual(result, expected) + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_usage_keys_to_remove_with_multiple_exclusions(self, mock_is_enabled: Mock): + """Test usage_keys_to_remove with multiple exclusions.""" + mock_is_enabled.return_value = True + # Create mock outline with multiple exclusions + section1 = MockOutlineSection( + "section1", + {USER_GROUP_PARTITION_ID: {3, 4}}, # User not in these groups + ) + section1.sequences = [MockOutlineSequence("seq1", {}), MockOutlineSequence("seq2", {})] + section2 = MockOutlineSection("section2", {}) + section2.sequences = [ + MockOutlineSequence( + "seq3", + {USER_GROUP_PARTITION_ID: {5, 6}}, # User not in these groups + ), + MockOutlineSequence("seq4", {}), + ] + full_course_outline = MockFullCourseOutline([section1, section2]) + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + processor.user_groups = [self.user_group_1, self.user_group_2] + result = processor.usage_keys_to_remove(full_course_outline) + + expected = {"section1", "seq1", "seq2", "seq3"} + self.assertEqual(result, expected) + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_usage_keys_to_remove_when_user_groups_disabled(self, mock_is_enabled: Mock): + """Test usage_keys_to_remove when user groups are disabled.""" + mock_is_enabled.return_value = False + # Create mock outline with exclusions + section1 = MockOutlineSection("section1", {USER_GROUP_PARTITION_ID: {3, 4}}) + section1.sequences = [MockOutlineSequence("seq1", {})] + full_course_outline = MockFullCourseOutline([section1]) + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + result = processor.usage_keys_to_remove(full_course_outline) + + self.assertEqual(result, set()) + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_usage_keys_to_remove_with_empty_outline(self, mock_is_enabled: Mock): + """Test usage_keys_to_remove with an empty course outline.""" + mock_is_enabled.return_value = True + full_course_outline = MockFullCourseOutline([]) + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + result = processor.usage_keys_to_remove(full_course_outline) + + self.assertEqual(result, set()) + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_usage_keys_to_remove_with_section_no_sequences(self, mock_is_enabled: Mock): + """Test usage_keys_to_remove with a section that has no sequences.""" + mock_is_enabled.return_value = True + # Create mock outline with section that has no sequences + section1 = MockOutlineSection( + "section1", + {USER_GROUP_PARTITION_ID: {3, 4}}, # User not in these groups + ) + section1.sequences = [] + full_course_outline = MockFullCourseOutline([section1]) + + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + processor.user_groups = [self.user_group_1, self.user_group_2] + result = processor.usage_keys_to_remove(full_course_outline) + + expected = {"section1"} + self.assertEqual(result, expected) + + @patch("openedx_user_groups.processors.user_group_partition_groups.is_user_groups_enabled") + def test_integration_load_data_and_exclusion_logic(self, mock_is_enabled: Mock): + """Integration test combining load_data and exclusion logic.""" + mock_is_enabled.return_value = True + processor = UserGroupPartitionGroupsOutlineProcessor(self.course_key, self.user, self.at_time) + # Mock the user groups that the user belongs to + processor.user_groups = [self.user_group_1, self.user_group_2] + + # Test exclusion logic with user in some groups but not others + result1 = processor._is_user_excluded_by_partition_group( + {USER_GROUP_PARTITION_ID: {1, 3}} # User in group 1, not in group 3 + ) + self.assertFalse(result1) # Should not be excluded because user is in group 1 + + result2 = processor._is_user_excluded_by_partition_group( + {USER_GROUP_PARTITION_ID: {3, 4}} # User not in either group + ) + self.assertTrue(result2) # Should be excluded because user is not in any group diff --git a/tests/test_user_group_partition_scheme.py b/tests/test_user_group_partition_scheme.py new file mode 100644 index 0000000..51035bc --- /dev/null +++ b/tests/test_user_group_partition_scheme.py @@ -0,0 +1,535 @@ +""" +Test Suite for the User Group Partition Scheme. + +This test suite covers all classes and methods defined in `user_group_partition_scheme.py`. +""" + +from unittest.mock import Mock, patch + +from django.contrib.auth import get_user_model +from django.contrib.contenttypes.models import ContentType +from django.test import TestCase +from opaque_keys.edx.keys import CourseKey + +from openedx_user_groups.models import Scope, UserGroup, UserGroupMembership +from openedx_user_groups.partitions.user_group_partition_scheme import ( + USER_GROUP_PARTITION_ID, + UserGroupPartition, + UserGroupPartitionScheme, + UserPartitionError, + create_user_group_partition, + create_user_group_partition_with_course_id, +) +from tests.factories import UserFactory + +User = get_user_model() + + +class GroupMock: + """Mock Group class for testing.""" + + def __init__(self, id, name): # pylint: disable=redefined-builtin + self.id = id + self.name = name + + def __eq__(self, other): + """Compare GroupMock instances by their attributes.""" + if not isinstance(other, GroupMock): + return False + return self.id == other.id and self.name == other.name + + +class UserPartitionMock: + """Mock UserPartition class for testing.""" + + # pylint: disable=redefined-builtin, too-many-positional-arguments + def __init__(self, id, name, description, groups, scheme, parameters, active=True): + self.id = id + self.name = name + self.description = description + self.groups = groups + self.scheme = scheme + self.parameters = parameters + self.active = active + + +class TestUserGroupPartition(TestCase): + """Test UserGroupPartition class.""" + + @classmethod + def setUpTestData(cls): + """Set up test data for UserGroupPartition tests.""" + cls.course_key = CourseKey.from_string("course-v1:OpenedX+Demo+Course") + cls.course_content_type = ContentType.objects.get_for_model(User) + + cls.scope = Scope.objects.create( + name="Demo Course Scope", + description="Scope for the demo course", + content_type=cls.course_content_type, + object_id=1, + ) + + cls.user_group1 = UserGroup.objects.create( + name="Test Group 1", + description="First test group", + scope=cls.scope, + enabled=True, + ) + + cls.user_group2 = UserGroup.objects.create( + name="Test Group 2", + description="Second test group", + scope=cls.scope, + enabled=True, + ) + + cls.disabled_user_group = UserGroup.objects.create( + name="Disabled Group", + description="Disabled test group", + scope=cls.scope, + enabled=False, + ) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.Group") + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + def test_groups_property_when_enabled(self, mock_is_enabled: Mock, mock_group: Mock): + """Test groups property returns enabled user groups when feature is enabled.""" + mock_is_enabled.return_value = True + mock_group.side_effect = GroupMock + + partition = UserGroupPartition( + id=USER_GROUP_PARTITION_ID, + name="Test Partition", + description="Test partition description", + groups=[], + scheme=UserGroupPartitionScheme, + parameters={"course_id": str(self.course_key)}, + ) + + groups = partition.groups + # Should return only enabled user groups + expected_group_ids = {self.user_group1.id, self.user_group2.id} + expected_group_names = {self.user_group1.name, self.user_group2.name} + + actual_group_ids = {group.id for group in groups} + actual_group_names = {group.name for group in groups} + + self.assertEqual(actual_group_ids, expected_group_ids) + self.assertEqual(actual_group_names, expected_group_names) + self.assertNotIn(self.disabled_user_group.id, actual_group_ids) + self.assertNotIn(self.disabled_user_group.name, actual_group_names) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + def test_groups_property_when_disabled(self, mock_is_enabled: Mock): + """Test groups property returns empty list when feature is disabled.""" + mock_is_enabled.return_value = False + + partition = UserGroupPartition( + id=USER_GROUP_PARTITION_ID, + name="Test Partition", + description="Test partition description", + groups=[], + scheme=UserGroupPartitionScheme, + parameters={"course_id": str(self.course_key)}, + ) + + groups = partition.groups + self.assertEqual(groups, []) + + +class TestUserGroupPartitionScheme(TestCase): + """Test UserGroupPartitionScheme class.""" + + @classmethod + def setUpTestData(cls): + """Set up test data for UserGroupPartitionScheme tests.""" + cls.course_key = CourseKey.from_string("course-v1:OpenedX+Demo+Course") + cls.course_content_type = ContentType.objects.get_for_model(User) + + cls.scope = Scope.objects.create( + name="Demo Course Scope", + description="Scope for the demo course", + content_type=cls.course_content_type, + object_id=1, + ) + + cls.user_group1 = UserGroup.objects.create( + name="Test Group 1", + description="First test group", + scope=cls.scope, + enabled=True, + ) + + cls.user_group2 = UserGroup.objects.create( + name="Test Group 2", + description="Second test group", + scope=cls.scope, + enabled=True, + ) + + cls.user = UserFactory() + + # Create user group memberships + cls.membership1 = UserGroupMembership.objects.create( + user=cls.user, + group=cls.user_group1, + is_active=True, + ) + + cls.membership2 = UserGroupMembership.objects.create( + user=cls.user, + group=cls.user_group2, + is_active=True, + ) + + def setUp(self): + """Set up for each test method.""" + self.user_partition = UserPartitionMock( + id=USER_GROUP_PARTITION_ID, + name="Test Partition", + description="Test partition description", + groups=[], + scheme=UserGroupPartitionScheme, + parameters={"course_id": str(self.course_key)}, + ) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.Group") + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + def test_get_group_for_user_when_enabled(self, mock_is_enabled: Mock, mock_group: Mock): + """Test get_group_for_user returns user groups when feature is enabled.""" + mock_is_enabled.return_value = True + mock_group.side_effect = GroupMock + + groups = UserGroupPartitionScheme.get_group_for_user(self.course_key, self.user, self.user_partition) + + # Should return groups for both user groups the user belongs to + self.assertIsNotNone(groups) + self.assertEqual(len(groups), 2) + + group_ids = [group.id for group in groups] + group_names = [group.name for group in groups] + + self.assertIn(self.user_group1.id, group_ids) + self.assertIn(self.user_group2.id, group_ids) + self.assertIn(self.user_group1.name, group_names) + self.assertIn(self.user_group2.name, group_names) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + def test_get_group_for_user_when_disabled(self, mock_is_enabled: Mock): + """Test get_group_for_user returns None when feature is disabled.""" + mock_is_enabled.return_value = False + + groups = UserGroupPartitionScheme.get_group_for_user(self.course_key, self.user, self.user_partition) + + self.assertIsNone(groups) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + def test_get_group_for_user_no_memberships(self, mock_is_enabled: Mock): + """Test get_group_for_user returns None when user has no memberships.""" + mock_is_enabled.return_value = True + # Create user with no memberships + user_without_memberships = UserFactory() + + groups = UserGroupPartitionScheme.get_group_for_user( + self.course_key, user_without_memberships, self.user_partition + ) + + self.assertIsNone(groups) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.Group") + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + def test_get_group_for_user_inactive_memberships(self, mock_is_enabled: Mock, mock_group: Mock): + """Test get_group_for_user ignores inactive memberships.""" + mock_is_enabled.return_value = True + mock_group.side_effect = GroupMock + # Deactivate one membership + self.membership2.is_active = False + self.membership2.save() + + groups = UserGroupPartitionScheme.get_group_for_user(self.course_key, self.user, self.user_partition) + + # Should only return the active membership + self.assertIsNotNone(groups) + self.assertEqual(len(groups), 1) + self.assertEqual(groups[0].id, self.user_group1.id) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + @patch("openedx_user_groups.partitions.user_group_partition_scheme.get_course_masquerade") + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_masquerading_as_specific_student") + @patch("openedx_user_groups.partitions.user_group_partition_scheme.get_masquerading_user_group") + def test_get_group_for_user_with_masquerade( + self, + mock_get_masquerading_group: Mock, + mock_is_masquerading: Mock, + mock_get_masquerade: Mock, + mock_is_enabled: Mock, + ): + """Test get_group_for_user handles masquerading correctly.""" + mock_is_enabled.return_value = True + mock_get_masquerade.return_value = True + mock_is_masquerading.return_value = False + expected_groups = [GroupMock(id=1, name="Masquerade Group")] + mock_get_masquerading_group.return_value = expected_groups + + groups = UserGroupPartitionScheme.get_group_for_user(self.course_key, self.user, self.user_partition) + + self.assertEqual(groups, expected_groups) + mock_get_masquerading_group.assert_called_once_with(self.course_key, self.user, self.user_partition) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + def test_create_user_partition_when_enabled(self, mock_is_enabled: Mock): + """Test create_user_partition creates partition when feature is enabled.""" + mock_is_enabled.return_value = True + + partition = UserGroupPartitionScheme.create_user_partition( + id=USER_GROUP_PARTITION_ID, + name="Test Partition", + description="Test partition description", + parameters={"course_id": str(self.course_key)}, + ) + + self.assertIsInstance(partition, UserGroupPartition) + self.assertEqual(partition.id, USER_GROUP_PARTITION_ID) + self.assertEqual(partition.name, "Test Partition") + self.assertEqual(partition.description, "Test partition description") + self.assertEqual(partition.scheme, UserGroupPartitionScheme) + self.assertEqual(partition.parameters, {"course_id": str(self.course_key)}) + self.assertTrue(partition.active) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + def test_create_user_partition_when_disabled(self, mock_is_enabled: Mock): + """Test create_user_partition returns None when feature is disabled.""" + mock_is_enabled.return_value = False + + partition = UserGroupPartitionScheme.create_user_partition( + id=USER_GROUP_PARTITION_ID, + name="Test Partition", + description="Test partition description", + parameters={"course_id": str(self.course_key)}, + ) + + self.assertIsNone(partition) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.Group") + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + def test_create_user_partition_with_groups(self, mock_is_enabled: Mock, mock_group: Mock): + """Test create_user_partition with provided groups.""" + mock_is_enabled.return_value = True + mock_group.side_effect = GroupMock + test_groups = [GroupMock(id=1, name="Test Group 1"), GroupMock(id=2, name="Test Group 2")] + + partition = UserGroupPartitionScheme.create_user_partition( + id=USER_GROUP_PARTITION_ID, + name="Test Partition", + description="Test partition description", + groups=test_groups, + parameters={"course_id": str(self.course_key)}, + active=False, + ) + + self.assertEqual(partition.groups, test_groups) + self.assertFalse(partition.active) + + +class TestPartitionCreationFunctions(TestCase): + """Test the partition creation helper functions.""" + + @classmethod + def setUpTestData(cls): + """Set up test data for partition creation function tests.""" + cls.course_key = CourseKey.from_string("course-v1:OpenedX+Demo+Course") + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + @patch("openedx_user_groups.partitions.user_group_partition_scheme.UserPartition") + def test_create_user_group_partition_with_course_id_success(self, mock_user_partition: Mock, mock_is_enabled: Mock): + """Test create_user_group_partition_with_course_id creates partition successfully.""" + mock_is_enabled.return_value = True + + mock_scheme = Mock() + mock_scheme.create_user_partition.return_value = Mock() + mock_user_partition.get_scheme.return_value = mock_scheme + + partition = create_user_group_partition_with_course_id(self.course_key) + + self.assertIsNotNone(partition) + mock_scheme.create_user_partition.assert_called_once_with( + id=USER_GROUP_PARTITION_ID, + name="User Groups", + description="Partition for segmenting users by user groups", + parameters={"course_id": str(self.course_key)}, + ) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.UserPartition") + def test_create_user_group_partition_with_course_id_scheme_not_found(self, mock_user_partition: Mock): + """Test create_user_group_partition_with_course_id handles missing scheme gracefully.""" + mock_user_partition.get_scheme.side_effect = UserPartitionError("Scheme not found") + + partition = create_user_group_partition_with_course_id(self.course_key) + + self.assertIsNone(partition) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.create_user_group_partition_with_course_id") + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + def test_create_user_group_partition_success(self, mock_is_enabled: Mock, mock_create_partition: Mock): + """Test create_user_group_partition returns partition when feature is enabled.""" + mock_is_enabled.return_value = True + course = Mock(id=self.course_key) + + partition = create_user_group_partition(course) + + self.assertIsNotNone(partition) + mock_create_partition.assert_called_once_with(self.course_key) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + def test_create_user_group_partition_disabled(self, mock_is_enabled: Mock): + """Test create_user_group_partition returns empty list when feature is disabled.""" + mock_is_enabled.return_value = False + course = Mock(id=self.course_key) + + partition = create_user_group_partition(course) + + self.assertEqual(partition, []) + + +class TestUserGroupPartitionSchemeIntegration(TestCase): + """Integration tests for UserGroupPartitionScheme with real database operations.""" + + @classmethod + def setUpTestData(cls): + """Set up test data for integration tests.""" + cls.course_key = CourseKey.from_string("course-v1:OpenedX+Demo+Course") + cls.course_content_type = ContentType.objects.get_for_model(User) + + cls.scope = Scope.objects.create( + name="Demo Course Scope", + description="Scope for the demo course", + content_type=cls.course_content_type, + object_id=1, + ) + + # Create multiple user groups + cls.user_group1 = UserGroup.objects.create( + name="Group A", + description="First group", + scope=cls.scope, + enabled=True, + ) + + cls.user_group2 = UserGroup.objects.create( + name="Group B", + description="Second group", + scope=cls.scope, + enabled=True, + ) + + cls.user_group3 = UserGroup.objects.create( + name="Group C", + description="Third group", + scope=cls.scope, + enabled=False, # Disabled group + ) + + # Create users + cls.user1 = UserFactory() + cls.user2 = UserFactory() + cls.user3 = UserFactory() + + # Create memberships + UserGroupMembership.objects.create( + user=cls.user1, + group=cls.user_group1, + is_active=True, + ) + + UserGroupMembership.objects.create( + user=cls.user2, + group=cls.user_group2, + is_active=True, + ) + + UserGroupMembership.objects.create( + user=cls.user3, + group=cls.user_group3, + is_active=True, + ) + + def setUp(self): + """Set up for each test method.""" + self.user_partition = UserPartitionMock( + id=USER_GROUP_PARTITION_ID, + name="Test Partition", + description="Test partition description", + groups=[], + scheme=UserGroupPartitionScheme, + parameters={"course_id": str(self.course_key)}, + ) + + def _assert_single_group(self, groups: list[GroupMock], expected_group: GroupMock): + """Helper method to assert a single group assignment.""" + self.assertIsNotNone(groups) + self.assertEqual(len(groups), 1) + self.assertEqual(groups[0].id, expected_group.id) + self.assertEqual(groups[0].name, expected_group.name) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.Group") + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + def test_integration_user_group_assignment(self, mock_is_enabled: Mock, mock_group: Mock): + """Test that users are correctly assigned to their user groups.""" + mock_is_enabled.return_value = True + mock_group.side_effect = GroupMock + + # Test user1 should be in Group A + groups1 = UserGroupPartitionScheme.get_group_for_user(self.course_key, self.user1, self.user_partition) + self._assert_single_group(groups1, self.user_group1) + + # Test user2 should be in Group B + groups2 = UserGroupPartitionScheme.get_group_for_user(self.course_key, self.user2, self.user_partition) + self._assert_single_group(groups2, self.user_group2) + + # Test user3 should not be in any group (since Group C is disabled) + groups3 = UserGroupPartitionScheme.get_group_for_user(self.course_key, self.user3, self.user_partition) + self.assertEqual(groups3, []) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.Group") + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + def test_integration_multiple_memberships(self, mock_is_enabled: Mock, mock_group: Mock): + """Test that users with multiple memberships get all their groups.""" + mock_is_enabled.return_value = True + mock_group.side_effect = GroupMock + + # Add user1 to Group B as well + UserGroupMembership.objects.create( + user=self.user1, + group=self.user_group2, + is_active=True, + ) + + groups = UserGroupPartitionScheme.get_group_for_user(self.course_key, self.user1, self.user_partition) + self.assertIsNotNone(groups) + self.assertEqual(len(groups), 2) + expected_group_ids = {self.user_group1.id, self.user_group2.id} + actual_group_ids = {group.id for group in groups} + self.assertEqual(actual_group_ids, expected_group_ids) + + @patch("openedx_user_groups.partitions.user_group_partition_scheme.Group") + @patch("openedx_user_groups.partitions.user_group_partition_scheme.is_user_groups_enabled") + def test_integration_partition_groups_property(self, mock_is_enabled: Mock, mock_group: Mock): + """Test that partition groups property returns only enabled groups.""" + mock_is_enabled.return_value = True + mock_group.side_effect = GroupMock + + partition = UserGroupPartition( + id=USER_GROUP_PARTITION_ID, + name="Test Partition", + description="Test partition description", + groups=[], + scheme=UserGroupPartitionScheme, + parameters={"course_id": str(self.course_key)}, + ) + + groups = partition.groups + # Should only return enabled groups + expected_group_ids = {self.user_group1.id, self.user_group2.id} + actual_group_ids = {group.id for group in groups} + self.assertEqual(actual_group_ids, expected_group_ids) + self.assertNotIn(self.user_group3.id, actual_group_ids) # Disabled group should not be included