def test_preserves_existing_user_partitions(self):
        # Add other, non-verified partition to the course
        self.course.user_partitions = [
            UserPartition(
                id=0,
                name='Cohort user partition',
                scheme=UserPartition.get_scheme('cohort'),
                description='Cohorted user partition',
                groups=[
                    Group(id=0, name="Group A"),
                    Group(id=1, name="Group B"),
                ],
            ),
            UserPartition(
                id=1,
                name='Random user partition',
                scheme=UserPartition.get_scheme('random'),
                description='Random user partition',
                groups=[
                    Group(id=0, name="Group A"),
                    Group(id=1, name="Group B"),
                ],
            ),
        ]
        self.course = self.store.update_item(self.course, ModuleStoreEnum.UserID.test)

        # Update the verification partitions.
        # The existing partitions should still be available
        self._update_partitions()
        partition_ids = [p.id for p in self.course.user_partitions]
        self.assertEqual(len(partition_ids), 3)
        self.assertIn(0, partition_ids)
        self.assertIn(1, partition_ids)

    def get_or_create_content_group(store, course):
        """
        Returns the first user partition from the course which uses the
        CohortPartitionScheme, or generates one if no such partition is
        found.  The created partition is not saved to the course until
        the client explicitly creates a group within the partition and
        POSTs back.
        """
        content_group_configuration = get_cohorted_user_partition(course)
        if content_group_configuration is None:
            content_group_configuration = UserPartition(
                id=generate_int_id(MINIMUM_GROUP_ID, MYSQL_MAX_INT, GroupConfiguration.get_used_ids(course)),
                name=CONTENT_GROUP_CONFIGURATION_NAME,
                description=CONTENT_GROUP_CONFIGURATION_DESCRIPTION,
                groups=[],
                scheme_id=COHORT_SCHEME
            )
            return content_group_configuration.to_json()

        content_group_configuration = GroupConfiguration.update_content_group_usage_info(
            store,
            course,
            content_group_configuration
        )
        return content_group_configuration

    def setUp(self):
        """Create a dummy course. """
        super(GetUserPartitionInfoTest, self).setUp()
        self.course = CourseFactory()
        self.block = ItemFactory.create(category="problem", parent_location=self.course.location)  # pylint: disable=no-member

        # Set up some default partitions
        self._set_partitions([
            UserPartition(
                id=0,
                name="Cohort user partition",
                scheme=UserPartition.get_scheme("cohort"),
                description="Cohorted user partition",
                groups=[
                    Group(id=0, name="Group A"),
                    Group(id=1, name="Group B"),
                ],
            ),
            UserPartition(
                id=1,
                name="Random user partition",
                scheme=UserPartition.get_scheme("random"),
                description="Random user partition",
                groups=[
                    Group(id=0, name="Group C"),
                ],
            ),
        ])

    def test_exclude_inactive_partitions(self):
        # Include an inactive verification scheme
        self._set_partitions([
            UserPartition(
                id=0,
                name="Cohort user partition",
                scheme=UserPartition.get_scheme("cohort"),
                description="Cohorted user partition",
                groups=[
                    Group(id=0, name="Group A"),
                    Group(id=1, name="Group B"),
                ],
            ),
            UserPartition(
                id=1,
                name="Verification user partition",
                scheme=UserPartition.get_scheme("verification"),
                description="Verification user partition",
                groups=[
                    Group(id=0, name="Group C"),
                ],
                active=False,
            ),
        ])

        # Expect that the inactive scheme is excluded from the results
        partitions = self._get_partition_info()
        self.assertEqual(len(partitions), 1)
        self.assertEqual(partitions[0]["scheme"], "cohort")

    def test_exclude_partitions_with_no_groups(self):
        # The cohort partition has no groups defined
        self._set_partitions([
            UserPartition(
                id=0,
                name="Cohort user partition",
                scheme=UserPartition.get_scheme("cohort"),
                description="Cohorted user partition",
                groups=[],
            ),
            UserPartition(
                id=1,
                name="Verification user partition",
                scheme=UserPartition.get_scheme("verification"),
                description="Verification user partition",
                groups=[
                    Group(id=0, name="Group C"),
                ],
            ),
        ])

        # Expect that the partition with no groups is excluded from the results
        partitions = self._get_partition_info()
        self.assertEqual(len(partitions), 1)
        self.assertEqual(partitions[0]["scheme"], "verification")

    def test_from_json_broken(self):
        # Missing field
        jsonified = {
            "name": self.TEST_NAME,
            "description": self.TEST_DESCRIPTION,
            "groups": [group.to_json() for group in self.TEST_GROUPS],
            "version": UserPartition.VERSION,
            "scheme": self.TEST_SCHEME_NAME,
        }
        with self.assertRaisesRegexp(TypeError, "missing value key 'id'"):
            UserPartition.from_json(jsonified)

        # Missing scheme
        jsonified = {
            'id': self.TEST_ID,
            "name": self.TEST_NAME,
            "description": self.TEST_DESCRIPTION,
            "groups": [group.to_json() for group in self.TEST_GROUPS],
            "version": UserPartition.VERSION,
        }
        with self.assertRaisesRegexp(TypeError, "missing value key 'scheme'"):
            UserPartition.from_json(jsonified)

        # Invalid scheme
        jsonified = {
            'id': self.TEST_ID,
            "name": self.TEST_NAME,
            "description": self.TEST_DESCRIPTION,
            "groups": [group.to_json() for group in self.TEST_GROUPS],
            "version": UserPartition.VERSION,
            "scheme": "no_such_scheme",
        }
        with self.assertRaisesRegexp(UserPartitionError, "Unrecognized scheme"):
            UserPartition.from_json(jsonified)

        # Wrong version (it's over 9000!)
        # Wrong version (it's over 9000!)
        jsonified = {
            'id': self.TEST_ID,
            "name": self.TEST_NAME,
            "description": self.TEST_DESCRIPTION,
            "groups": [group.to_json() for group in self.TEST_GROUPS],
            "version": 9001,
            "scheme": self.TEST_SCHEME_NAME,
        }
        with self.assertRaisesRegexp(TypeError, "has unexpected version"):
            UserPartition.from_json(jsonified)

        # Has extra key - should not be a problem
        jsonified = {
            'id': self.TEST_ID,
            "name": self.TEST_NAME,
            "description": self.TEST_DESCRIPTION,
            "groups": [group.to_json() for group in self.TEST_GROUPS],
            "version": UserPartition.VERSION,
            "scheme": "mock",
            "programmer": "Cale",
        }
        user_partition = UserPartition.from_json(jsonified)
        self.assertNotIn("programmer", user_partition.to_json())

class PartitionTestCase(TestCase):
    """Base class for test cases that require partitions"""
    shard = 2
    TEST_ID = 0
    TEST_NAME = "Mock Partition"
    TEST_DESCRIPTION = "for testing purposes"
    TEST_PARAMETERS = {"location": "block-v1:[email protected]@uuid"}
    TEST_GROUPS = [Group(0, 'Group 1'), Group(1, 'Group 2')]
    TEST_SCHEME_NAME = "mock"
    ENROLLMENT_TRACK_SCHEME_NAME = "enrollment_track"

    def setUp(self):
        super(PartitionTestCase, self).setUp()
        # Set up two user partition schemes: mock and random
        self.non_random_scheme = MockUserPartitionScheme(self.TEST_SCHEME_NAME)
        self.random_scheme = MockUserPartitionScheme("random")
        self.enrollment_track_scheme = MockEnrollmentTrackUserPartitionScheme(self.ENROLLMENT_TRACK_SCHEME_NAME)
        extensions = [
            Extension(
                self.non_random_scheme.name, USER_PARTITION_SCHEME_NAMESPACE, self.non_random_scheme, None
            ),
            Extension(
                self.random_scheme.name, USER_PARTITION_SCHEME_NAMESPACE, self.random_scheme, None
            ),
            Extension(
                self.enrollment_track_scheme.name, USER_PARTITION_SCHEME_NAMESPACE, self.enrollment_track_scheme, None
            ),
        ]
        UserPartition.scheme_extensions = ExtensionManager.make_test_instance(
            extensions, namespace=USER_PARTITION_SCHEME_NAMESPACE
        )

        # Be sure to clean up the global scheme_extensions after the test.
        self.addCleanup(self.cleanup_scheme_extensions)

        # Create a test partition
        self.user_partition = UserPartition(
            self.TEST_ID,
            self.TEST_NAME,
            self.TEST_DESCRIPTION,
            self.TEST_GROUPS,
            extensions[0].plugin,
            self.TEST_PARAMETERS,
        )

        # Make sure the names are set on the schemes (which happens normally in code, but may not happen in tests).
        self.user_partition.get_scheme(self.non_random_scheme.name)
        self.user_partition.get_scheme(self.random_scheme.name)

    def cleanup_scheme_extensions(self):
        """
        Unset the UserPartition.scheme_extensions cache.
        """
        UserPartition.scheme_extensions = None

    def test_to_json(self):
        groups = [Group(0, 'Group 1'), Group(1, 'Group 2')]
        upid = 0
        upname = "Test Partition"
        updesc = "for testing purposes"
        user_partition = UserPartition(upid, upname, updesc, groups)

        jsonified = user_partition.to_json()
        act_jsonified = {
            "id": upid,
            "name": upname,
            "description": updesc,
            "groups": [group.to_json() for group in groups],
            "version": user_partition.VERSION
        }
        self.assertEqual(jsonified, act_jsonified)

    def test_group_access_short_circuits(self):
        """
        Test that the group_access check short-circuits if there are no user_partitions defined
        except user_partitions in use by the split_test module.
        """
        # Initially, "red_cat" user can't view the vertical.
        self.set_group_access(self.chapter_location, {self.animal_partition.id: [self.dog_group.id]})
        self.check_access(self.red_cat, self.vertical_location, False)

        # Change the vertical's user_partitions value to the empty list. Now red_cat can view the vertical.
        self.set_user_partitions(self.vertical_location, [])
        self.check_access(self.red_cat, self.vertical_location, True)

        # Change the vertical's user_partitions value to include only "split_test" partitions.
        split_test_partition = UserPartition(
            199,
            'split_test partition',
            'nothing to look at here',
            [Group(2, 'random group')],
            scheme=UserPartition.get_scheme("random"),
        )
        self.set_user_partitions(self.vertical_location, [split_test_partition])
        self.check_access(self.red_cat, self.vertical_location, True)

        # Finally, add back in a cohort user_partition
        self.set_user_partitions(self.vertical_location, [split_test_partition, self.animal_partition])
        self.check_access(self.red_cat, self.vertical_location, False)

    def add_seq_with_content_groups(self, groups=None):
        """
        Adds sequential and two content groups to first course in courses list.
        """
        config_course_cohorts(self.courses[0], is_cohorted=True)

        if groups is None:
            groups = self.groups

        self.user_partition = UserPartition(
            id=0,
            name='Partition 1',
            description='This is partition 1',
            groups=groups,
            scheme=CohortPartitionScheme
        )

        self.user_partition.scheme.name = "cohort"

        ItemFactory.create(
            parent_location=self.chapter.location,
            category='sequential',
            display_name="Lesson 1",
            publish_item=True,
            metadata={u"user_partitions": [self.user_partition.to_json()]}
        )

        self.first_cohort, self.second_cohort = [
            CohortFactory(course_id=self.courses[0].id) for _ in range(2)
        ]

        self.courses[0].user_partitions = [self.user_partition]
        self.courses[0].save()
        modulestore().update_item(self.courses[0], self.user.id)

    def setUp(self):
        super(PartitionTestCase, self).setUp()
        # Set up two user partition schemes: mock and random
        self.non_random_scheme = MockUserPartitionScheme(self.TEST_SCHEME_NAME)
        self.random_scheme = MockUserPartitionScheme("random")
        extensions = [
            Extension(
                self.non_random_scheme.name, USER_PARTITION_SCHEME_NAMESPACE, self.non_random_scheme, None
            ),
            Extension(
                self.random_scheme.name, USER_PARTITION_SCHEME_NAMESPACE, self.random_scheme, None
            ),
        ]
        UserPartition.scheme_extensions = ExtensionManager.make_test_instance(
            extensions, namespace=USER_PARTITION_SCHEME_NAMESPACE
        )

        # Create a test partition
        self.user_partition = UserPartition(
            self.TEST_ID,
            self.TEST_NAME,
            self.TEST_DESCRIPTION,
            self.TEST_GROUPS,
            extensions[0].plugin,
            self.TEST_PARAMETERS,
        )

        # Make sure the names are set on the schemes (which happens normally in code, but may not happen in tests).
        self.user_partition.get_scheme(self.non_random_scheme.name)
        self.user_partition.get_scheme(self.random_scheme.name)

    def setUp(self):
        super(ReverificationPartitionTest, self).setUp()

        # creating course, checkpoint location and user partition mock object.
        self.course = CourseFactory.create()
        self.checkpoint_location = u'i4x://{org}/{course}/edx-reverification-block/first_uuid'.format(
            org=self.course.id.org, course=self.course.id.course
        )

        scheme = UserPartition.get_scheme("verification")
        self.user_partition = UserPartition(
            id=0,
            name=u"Verification Checkpoint",
            description=u"Verification Checkpoint",
            scheme=scheme,
            parameters={"location": self.checkpoint_location},
            groups=[
                Group(scheme.ALLOW, "Allow access to content"),
                Group(scheme.DENY, "Deny access to content"),
            ]
        )

        self.first_checkpoint = VerificationCheckpoint.objects.create(
            course_id=self.course.id,
            checkpoint_location=self.checkpoint_location
        )

def _create_enrollment_track_partition(course):
    """
    Create and return the dynamic enrollment track user partition.
    If it cannot be created, None is returned.
    """
    if not FEATURES.get('ENABLE_ENROLLMENT_TRACK_USER_PARTITION'):
        return None

    try:
        enrollment_track_scheme = UserPartition.get_scheme("enrollment_track")
    except UserPartitionError:
        log.warning("No 'enrollment_track' scheme registered, EnrollmentTrackUserPartition will not be created.")
        return None

    used_ids = set(p.id for p in course.user_partitions)
    if ENROLLMENT_TRACK_PARTITION_ID in used_ids:
        log.warning(
            "Can't add 'enrollment_track' partition, as ID {id} is assigned to {partition} in course {course}.".format(
                id=ENROLLMENT_TRACK_PARTITION_ID,
                partition=_get_partition_from_id(course.user_partitions, ENROLLMENT_TRACK_PARTITION_ID).name,
                course=unicode(course.id)
            )
        )
        return None

    partition = enrollment_track_scheme.create_user_partition(
        id=ENROLLMENT_TRACK_PARTITION_ID,
        name=_(u"Enrollment Track Groups"),
        description=_(u"Partition for segmenting users by enrollment track"),
        parameters={"course_id": unicode(course.id)}
    )
    return partition

 def get_user_partition(self):
     """
     Get user partition for saving in course.
     """
     try:
         return UserPartition.from_json(self.configuration)
     except ReadOnlyUserPartitionError:
         raise GroupConfigurationsValidationError(_("unable to load this type of group configuration"))

    def test_version_upgrade_2_to_3(self):
        # Test that version 3 user partition raises error if 'scheme' field is
        # not provided (same behavior as version 2)
        jsonified = {
            'id': self.TEST_ID,
            "name": self.TEST_NAME,
            "description": self.TEST_DESCRIPTION,
            "parameters": self.TEST_PARAMETERS,
            "groups": [group.to_json() for group in self.TEST_GROUPS],
            "version": 2,
        }
        with self.assertRaisesRegexp(TypeError, "missing value key 'scheme'"):
            UserPartition.from_json(jsonified)

        # Test that version 3 partitions have a scheme specified
        # and a field 'parameters' (optional while setting user partition but
        # always present in response)
        jsonified = {
            "id": self.TEST_ID,
            "name": self.TEST_NAME,
            "description": self.TEST_DESCRIPTION,
            "groups": [group.to_json() for group in self.TEST_GROUPS],
            "version": 2,
            "scheme": self.TEST_SCHEME_NAME,
        }
        user_partition = UserPartition.from_json(jsonified)
        self.assertEqual(user_partition.scheme.name, self.TEST_SCHEME_NAME)
        self.assertEqual(user_partition.parameters, {})
        self.assertTrue(user_partition.active)

        # now test that parameters dict is present in response with same value
        # as provided
        jsonified = {
            "id": self.TEST_ID,
            "name": self.TEST_NAME,
            "description": self.TEST_DESCRIPTION,
            "groups": [group.to_json() for group in self.TEST_GROUPS],
            "parameters": self.TEST_PARAMETERS,
            "version": 3,
            "scheme": self.TEST_SCHEME_NAME,
        }
        user_partition = UserPartition.from_json(jsonified)
        self.assertEqual(user_partition.parameters, self.TEST_PARAMETERS)
        self.assertTrue(user_partition.active)

    def test_create_user_partition(self):
        user_partition = UserPartition.get_scheme('enrollment_track').create_user_partition(
            301, "partition", "test partition", parameters={"course_id": unicode(self.course.id)}
        )
        self.assertEqual(type(user_partition), EnrollmentTrackUserPartition)
        self.assertEqual(user_partition.name, "partition")

        groups = user_partition.groups
        self.assertEqual(1, len(groups))
        self.assertEqual("Audit", groups[0].name)

 def test_version_upgrade(self):
     # Version 1 partitions did not have a scheme specified
     jsonified = {
         "id": self.TEST_ID,
         "name": self.TEST_NAME,
         "description": self.TEST_DESCRIPTION,
         "groups": [group.to_json() for group in self.TEST_GROUPS],
         "version": 1,
     }
     user_partition = UserPartition.from_json(jsonified)
     self.assertEqual(user_partition.scheme.name, "random")

    def test_can_handle_duplicate_group_ids(self):
        # Create the user partitions
        self.course.user_partitions = [
            UserPartition(
                id=0,
                name='Cohort user partition 1',
                scheme=UserPartition.get_scheme('cohort'),
                description='Cohorted user partition',
                groups=[
                    Group(id=2, name="Group 1A"),
                    Group(id=3, name="Group 1B"),
                ],
            ),
            UserPartition(
                id=1,
                name='Cohort user partition 2',
                scheme=UserPartition.get_scheme('cohort'),
                description='Random user partition',
                groups=[
                    Group(id=2, name="Group 2A"),
                    Group(id=3, name="Group 2B"),
                ],
            ),
        ]
        self.store.update_item(self.course, ModuleStoreEnum.UserID.test)

        # Assign group access rules for multiple partitions, one of which is a cohorted partition
        self._create_problem_with_content_group(0, 2, name_suffix='0')
        self._create_problem_with_content_group(1, 3, name_suffix='1')

        # This used to cause an exception since the code assumed that
        # only one partition would be available.
        actual = GroupConfiguration.get_partitions_usage_info(self.store, self.course)
        self.assertEqual(actual.keys(), [0, 1])
        self.assertEqual(actual[0].keys(), [2])
        self.assertEqual(actual[1].keys(), [3])

        actual = GroupConfiguration.get_content_groups_items_usage_info(self.store, self.course)
        self.assertEqual(actual.keys(), [0, 1])
        self.assertEqual(actual[0].keys(), [2])
        self.assertEqual(actual[1].keys(), [3])

    def test_can_handle_multiple_partitions(self):
        # Create the user partitions
        self.course.user_partitions = [
            UserPartition(
                id=0,
                name='Cohort user partition',
                scheme=UserPartition.get_scheme('cohort'),
                description='Cohorted user partition',
                groups=[
                    Group(id=0, name="Group A"),
                    Group(id=1, name="Group B"),
                ],
            ),
            UserPartition(
                id=1,
                name='Random user partition',
                scheme=UserPartition.get_scheme('random'),
                description='Random user partition',
                groups=[
                    Group(id=0, name="Group A"),
                    Group(id=1, name="Group B"),
                ],
            ),
        ]
        self.store.update_item(self.course, ModuleStoreEnum.UserID.test)

        # Assign group access rules for multiple partitions, one of which is a cohorted partition
        __, problem = self._create_problem_with_content_group(0, 1)
        problem.group_access = {
            0: [0],
            1: [1],
        }
        self.store.update_item(problem, ModuleStoreEnum.UserID.test)

        # This used to cause an exception since the code assumed that
        # only one partition would be available.
        actual = GroupConfiguration.get_content_groups_usage_info(self.store, self.course)
        self.assertEqual(actual.keys(), [0])

        actual = GroupConfiguration.get_content_groups_items_usage_info(self.store, self.course)
        self.assertEqual(actual.keys(), [0])

def create_enrollment_track_partition(course):
    """
    Create an EnrollmentTrackUserPartition instance for the given course.
    """
    enrollment_track_scheme = UserPartition.get_scheme("enrollment_track")
    partition = enrollment_track_scheme.create_user_partition(
        id=1,
        name="Test Enrollment Track Partition",
        description="Test partition for segmenting users by enrollment track",
        parameters={"course_id": unicode(course.id)}
    )
    return partition