def test_add_milestone_inactive_milestone_with_relationships_propagate_false(self):
        """ Unit Test: test_add_milestone_inactive_milestone_with_relationships_propagate_false"""
        milestone_data = {
            'name': 'local_milestone',
            'display_name': 'Local Milestone',
            'namespace': six.text_type(self.test_course_key),
            'description': 'Local Milestone Description'
        }
        milestone = api.add_milestone(milestone_data)
        api.add_course_milestone(
            self.test_course_key,
            self.relationship_types['REQUIRES'],
            milestone
        )
        api.add_course_content_milestone(
            self.test_course_key,
            self.test_content_key,
            self.relationship_types['FULFILLS'],
            milestone
        )
        api.add_user_milestone(self.serialized_test_user, milestone)
        self.assertGreater(milestone['id'], 0)
        api.remove_milestone(milestone['id'])

        with self.assertNumQueries(3):
            milestone = api.add_milestone(milestone_data, propagate=False)

 def test_remove_user_milestone(self):
     """ Unit Test: test_remove_user_milestone """
     api.add_user_milestone(self.serialized_test_user, self.test_milestone)
     self.assertTrue(api.user_has_milestone(self.serialized_test_user, self.test_milestone))
     with self.assertNumQueries(2):
         api.remove_user_milestone(self.serialized_test_user, self.test_milestone)
     self.assertFalse(api.user_has_milestone(self.serialized_test_user, self.test_milestone))

 def test_add_user_milestone_inactive_to_active(self):
     """ Unit Test: test_add_user_milestone """
     api.add_user_milestone(self.serialized_test_user, self.test_milestone)
     api.remove_user_milestone(self.serialized_test_user, self.test_milestone)
     with self.assertNumQueries(4):
         api.add_user_milestone(self.serialized_test_user, self.test_milestone)
     self.assertTrue(api.user_has_milestone(self.serialized_test_user, self.test_milestone))

 def _fulfill_content_milestones(user, course_key, content_key):
     """
     Internal helper to handle milestone fulfillments for the specified content module
     """
     # Fulfillment Use Case: Entrance Exam
     # If this module is part of an entrance exam, we'll need to see if the student
     # has reached the point at which they can collect the associated milestone
     if settings.FEATURES.get('ENTRANCE_EXAMS', False):
         course = modulestore().get_course(course_key)
         content = modulestore().get_item(content_key)
         entrance_exam_enabled = getattr(course, 'entrance_exam_enabled', False)
         in_entrance_exam = getattr(content, 'in_entrance_exam', False)
         if entrance_exam_enabled and in_entrance_exam:
             exam_pct = _calculate_entrance_exam_score(user, course)
             if exam_pct >= course.entrance_exam_minimum_score_pct:
                 exam_key = UsageKey.from_string(course.entrance_exam_id)
                 relationship_types = milestones_api.get_milestone_relationship_types()
                 content_milestones = milestones_api.get_course_content_milestones(
                     course_key,
                     exam_key,
                     relationship=relationship_types['FULFILLS']
                 )
                 # Add each milestone to the user's set...
                 user = {'id': user.id}
                 for milestone in content_milestones:
                     milestones_api.add_user_milestone(user, milestone)

 def test_add_user_milestone_bogus_user(self):
     """ Unit Test: test_add_user_milestone_bogus_user """
     with self.assertNumQueries(0):
         with self.assertRaises(exceptions.InvalidUserException):
             api.add_user_milestone({'identifier': 'abcd'}, self.test_milestone)
     with self.assertNumQueries(0):
         with self.assertRaises(exceptions.InvalidUserException):
             api.add_user_milestone(None, self.test_milestone)

def fulfill_course_milestone(course_key, user):
    """
    Marks the course specified by the given course_key as complete for the given user.
    If any other courses require this course as a prerequisite, their milestones will be appropriately updated.
    """
    if settings.FEATURES.get('MILESTONES_APP', False):
        course_milestones = get_course_milestones(course_key=course_key, relationship="fulfills")
        for milestone in course_milestones:
            add_user_milestone({'id': user.id}, milestone)

def evaluate_prerequisite(course, prereq_content_key, user_id):
    """
    Finds the parent subsection of the content in the course and evaluates
    any milestone relationships attached to that subsection. If the calculated
    grade of the prerequisite subsection meets the minimum score required by
    dependent subsections, the related milestone will be fulfilled for the user.

    Arguments:
        user_id (int): ID of User for which evaluation should occur
        course (CourseModule): The course
        prereq_content_key (UsageKey): The prerequisite content usage key

    Returns:
        None
    """
    xblock = modulestore().get_item(prereq_content_key)
    sequential = _get_xblock_parent(xblock, 'sequential')
    if sequential:
        prereq_milestone = gating_api.get_gating_milestone(
            course.id,
            sequential.location.for_branch(None),
            'fulfills'
        )
        if prereq_milestone:
            gated_content_milestones = defaultdict(list)
            for milestone in gating_api.find_gating_milestones(course.id, None, 'requires'):
                gated_content_milestones[milestone['id']].append(milestone)

            gated_content = gated_content_milestones.get(prereq_milestone['id'])
            if gated_content:
                from courseware.grades import get_module_score
                user = User.objects.get(id=user_id)
                score = get_module_score(user, course, sequential) * 100
                for milestone in gated_content:
                    # Default minimum score to 100
                    min_score = 100
                    requirements = milestone.get('requirements')
                    if requirements:
                        try:
                            min_score = int(requirements.get('min_score'))
                        except (ValueError, TypeError):
                            log.warning(
                                'Failed to find minimum score for gating milestone %s, defaulting to 100',
                                json.dumps(milestone)
                            )

                    if score >= min_score:
                        milestones_api.add_user_milestone({'id': user_id}, prereq_milestone)
                    else:
                        milestones_api.remove_user_milestone({'id': user_id}, prereq_milestone)

def fulfill_course_milestone(course_key, user):
    """
    Marks the course specified by the given course_key as complete for the given user.
    If any other courses require this course as a prerequisite, their milestones will be appropriately updated.
    """
    if not settings.FEATURES.get('MILESTONES_APP'):
        return None
    try:
        course_milestones = milestones_api.get_course_milestones(course_key=course_key, relationship="fulfills")
    except InvalidMilestoneRelationshipTypeException:
        # we have not seeded milestone relationship types
        seed_milestone_relationship_types()
        course_milestones = milestones_api.get_course_milestones(course_key=course_key, relationship="fulfills")
    for milestone in course_milestones:
        milestones_api.add_user_milestone({'id': user.id}, milestone)

def add_user_milestone(user, milestone):
    """
    Client API operation adapter/wrapper
    """
    if not settings.FEATURES.get('MILESTONES_APP'):
        return None
    return milestones_api.add_user_milestone(user, milestone)

 def _fulfill_content_milestones(course_key, content_key, user_id):  # pylint: disable=unused-argument
     """
     Internal helper to handle milestone fulfillments for the specified content module
     """
     # Fulfillment Use Case: Entrance Exam
     # If this module is part of an entrance exam, we'll need to see if the student
     # has reached the point at which they can collect the associated milestone
     if settings.FEATURES.get('ENTRANCE_EXAMS', False):
         course = modulestore().get_course(course_key)
         content = modulestore().get_item(content_key)
         entrance_exam_enabled = getattr(course, 'entrance_exam_enabled', False)
         in_entrance_exam = getattr(content, 'in_entrance_exam', False)
         if entrance_exam_enabled and in_entrance_exam:
             exam_key = UsageKey.from_string(course.entrance_exam_id)
             exam_descriptor = modulestore().get_item(exam_key)
             exam_modules = yield_dynamic_descriptor_descendents(
                 exam_descriptor,
                 inner_get_module
             )
             ignore_categories = ['course', 'chapter', 'sequential', 'vertical']
             module_pcts = []
             exam_pct = 0
             for module in exam_modules:
                 if module.graded and module.category not in ignore_categories:
                     module_pct = 0
                     try:
                         student_module = StudentModule.objects.get(
                             module_state_key=module.scope_ids.usage_id,
                             student_id=user_id
                         )
                         if student_module.max_grade:
                             module_pct = student_module.grade / student_module.max_grade
                     except StudentModule.DoesNotExist:
                         pass
                     module_pcts.append(module_pct)
             exam_pct = sum(module_pcts) / float(len(module_pcts))
             if exam_pct >= course.entrance_exam_minimum_score_pct:
                 relationship_types = milestones_api.get_milestone_relationship_types()
                 content_milestones = milestones_api.get_course_content_milestones(
                     course_key,
                     exam_key,
                     relationship=relationship_types['FULFILLS']
                 )
                 # Add each milestone to the user's set...
                 user = {'id': user_id}
                 for milestone in content_milestones:
                     milestones_api.add_user_milestone(user, milestone)

    def test_get_gated_content(self):
        """ Test test_get_gated_content """

        mock_user = MagicMock()
        mock_user.id.return_value = 1

        self.assertEqual(gating_api.get_gated_content(self.course, mock_user), [])

        gating_api.add_prerequisite(self.course.id, self.seq1.location)
        gating_api.set_required_content(self.course.id, self.seq2.location, self.seq1.location, 100)
        milestone = milestones_api.get_course_content_milestones(self.course.id, self.seq2.location, 'requires')[0]

        self.assertEqual(gating_api.get_gated_content(self.course, mock_user), [unicode(self.seq2.location)])

        milestones_api.add_user_milestone({'id': mock_user.id}, milestone)

        self.assertEqual(gating_api.get_gated_content(self.course, mock_user), [])

    def test_fetch_user_milestones_missing_match_criteria_throws_exception(self):
        """ Unit Test: test_fetch_user_milestones_missing_match_criteria_throws_exception """
        milestone1 = api.add_milestone({
            'display_name': 'Test Milestone',
            'name': 'test_milestone',
            'namespace': unicode(self.test_course_key),
            'description': 'Test Milestone Description',
        })
        api.add_user_milestone(self.serialized_test_user, milestone1)

        with self.assertRaises(exceptions.InvalidMilestoneException):
            data.fetch_user_milestones(self.serialized_test_user, {})

        # Ensure we cover all remaining logical branches per coverage.py
        data.fetch_user_milestones(
            self.serialized_test_user,
            {'id': milestone1['id']}
        )

 def test_get_course_content_milestones_with_fulfilled_user_milestones(self):
     """ Unit Test: test_get_course_content_milestones_with_fulfilled_user_milestones """
     user = {'id': self.test_user.id}
     api.add_course_content_milestone(
         self.test_course_key,
         self.test_content_key,
         self.relationship_types['REQUIRES'],
         self.test_milestone
     )
     api.add_user_milestone(user, self.test_milestone)
     with self.assertNumQueries(2):
         requirer_milestones = api.get_course_content_milestones(
             self.test_course_key,
             self.test_content_key,
             self.relationship_types['REQUIRES'],
             {'id': self.test_user.id}
         )
     self.assertEqual(len(requirer_milestones), 0)

    def test_get_gated_content(self):
        """
        Verify staff bypasses gated content and student gets list of unfulfilled prerequisites.
        """

        staff = UserFactory(is_staff=True)
        student = UserFactory(is_staff=False)

        self.assertEqual(gating_api.get_gated_content(self.course, staff), [])
        self.assertEqual(gating_api.get_gated_content(self.course, student), [])

        gating_api.add_prerequisite(self.course.id, self.seq1.location)
        gating_api.set_required_content(self.course.id, self.seq2.location, self.seq1.location, 100)
        milestone = milestones_api.get_course_content_milestones(self.course.id, self.seq2.location, 'requires')[0]

        self.assertEqual(gating_api.get_gated_content(self.course, staff), [])
        self.assertEqual(gating_api.get_gated_content(self.course, student), [unicode(self.seq2.location)])

        milestones_api.add_user_milestone({'id': student.id}, milestone)  # pylint: disable=no-member

        self.assertEqual(gating_api.get_gated_content(self.course, student), [])

#.........这里部分代码省略.........
            UsageKey.from_string('i4x://the/content/key/123456789'),
            self.relationship_types['FULFILLS'],
            local_milestone_2
        )
        api.add_course_content_milestone(
            self.test_course_key,
            UsageKey.from_string('i4x://the/content/key/123456789'),
            self.relationship_types['FULFILLS'],
            local_milestone_3
        )
        api.add_course_content_milestone(
            self.test_course_key,
            UsageKey.from_string('i4x://the/content/key/987654321'),
            self.relationship_types['FULFILLS'],
            local_milestone_3
        )

        # Confirm the starting state for this test (user has no milestones, course requires three)
        self.assertEqual(
            len(api.get_user_milestones(self.serialized_test_user, namespace=namespace)), 0)
        self.assertEqual(
            len(api.get_course_required_milestones(self.test_course_key, self.serialized_test_user)),
            3
        )
        # Check the possible fulfillment paths for the milestones for this course
        with self.assertNumQueries(8):
            paths = api.get_course_milestones_fulfillment_paths(
                self.test_course_key,
                self.serialized_test_user
            )

        # Set up the key values we'll use to access/assert the response
        milestone_key_1 = '{}.{}'.format(local_milestone_1['namespace'], local_milestone_1['name'])
        milestone_key_2 = '{}.{}'.format(local_milestone_2['namespace'], local_milestone_2['name'])
        milestone_key_3 = '{}.{}'.format(local_milestone_3['namespace'], local_milestone_3['name'])

        # First round of assertions
        self.assertEqual(len(paths[milestone_key_1]['courses']), 1)
        self.assertIsNone(paths[milestone_key_1].get('content'))
        self.assertEqual(len(paths[milestone_key_2]['courses']), 1)
        self.assertEqual(len(paths[milestone_key_2]['content']), 1)
        self.assertIsNone(paths[milestone_key_3].get('courses'))
        self.assertEqual(len(paths[milestone_key_3]['content']), 2)

        # Collect the first milestone (two should remain)
        api.add_user_milestone(self.serialized_test_user, local_milestone_1)
        self.assertEqual(
            len(api.get_user_milestones(self.serialized_test_user, namespace=namespace)), 1)
        self.assertEqual(
            len(api.get_course_required_milestones(self.test_course_key, self.serialized_test_user)),
            2
        )
        # Check the remaining fulfillment paths for the milestones for this course
        with self.assertNumQueries(6):
            paths = api.get_course_milestones_fulfillment_paths(
                self.test_course_key,
                self.serialized_test_user
            )
        self.assertIsNone(paths.get(milestone_key_1))
        self.assertEqual(len(paths[milestone_key_2]['courses']), 1)
        self.assertEqual(len(paths[milestone_key_2]['content']), 1)
        self.assertIsNone(paths[milestone_key_3].get('courses'))
        self.assertEqual(len(paths[milestone_key_3]['content']), 2)

        # Collect the second milestone (one should remain)
        api.add_user_milestone(self.serialized_test_user, local_milestone_2)
        self.assertEqual(
            len(api.get_user_milestones(self.serialized_test_user, namespace=namespace)), 2)
        self.assertEqual(
            len(api.get_course_required_milestones(self.test_course_key, self.serialized_test_user)),
            1
        )
        # Check the remaining fulfillment paths for the milestones for this course
        with self.assertNumQueries(4):
            paths = api.get_course_milestones_fulfillment_paths(
                self.test_course_key,
                self.serialized_test_user
            )
        self.assertIsNone(paths.get(milestone_key_1))
        self.assertIsNone(paths.get(milestone_key_2))
        self.assertIsNone(paths[milestone_key_3].get('courses'))
        self.assertEqual(len(paths[milestone_key_3]['content']), 2)

        # Collect the third milestone
        api.add_user_milestone(self.serialized_test_user, local_milestone_3)
        self.assertEqual(
            len(api.get_user_milestones(self.serialized_test_user, namespace=namespace)), 3)
        self.assertEqual(
            len(api.get_course_required_milestones(self.test_course_key, self.serialized_test_user)),
            0
        )
        # Check the remaining fulfillment paths for the milestones for this course
        with self.assertNumQueries(2):
            paths = api.get_course_milestones_fulfillment_paths(
                self.test_course_key,
                self.serialized_test_user
            )
        self.assertIsNone(paths.get(milestone_key_1))
        self.assertIsNone(paths.get(milestone_key_2))
        self.assertIsNone(paths.get(milestone_key_3))

    def test_get_course_unfulfilled_milestones(self):
        """ Unit Test: test_get_course_unfulfilled_milestones """
        namespace = 'test_get_milestones'
        milestone1 = api.add_milestone({
            'name': 'localmilestone1',
            'display_name': 'Local Milestone 1',
            'namespace': namespace,
            'description': 'Local Milestone 1 Description'
        })
        api.add_course_milestone(
            self.test_course_key,
            self.relationship_types['REQUIRES'],
            milestone1
        )

        milestone2 = api.add_milestone({
            'name': 'localmilestone2',
            'display_name': 'Local Milestone 2',
            'namespace': namespace,
            'description': 'Local Milestone 2 Description'
        })
        api.add_course_milestone(
            self.test_course_key,
            self.relationship_types['REQUIRES'],
            milestone2
        )

        # Confirm that the course has only two milestones, and that the User still needs to collect both
        course_milestones = api.get_course_milestones(self.test_course_key)
        self.assertEqual(len(course_milestones), 2)
        with self.assertNumQueries(2):
            required_milestones = api.get_course_required_milestones(
                self.test_course_key,
                self.serialized_test_user
            )

        # Link the User to Milestone 2 (this one is now 'collected')
        api.add_user_milestone(self.serialized_test_user, milestone2)
        user_milestones = api.get_user_milestones(self.serialized_test_user, namespace=namespace)
        self.assertEqual(len(user_milestones), 1)
        self.assertEqual(user_milestones[0]['id'], milestone2['id'])

        # Only Milestone 1 should be listed as 'required' for the course at this point
        with self.assertNumQueries(2):
            required_milestones = api.get_course_required_milestones(
                self.test_course_key,
                self.serialized_test_user
            )
        self.assertEqual(len(required_milestones), 1)
        self.assertEqual(required_milestones[0]['id'], milestone1['id'])

        # Link the User to Milestone 1 (this one is now 'collected', as well)
        api.add_user_milestone(self.serialized_test_user, milestone1)
        user_milestones = api.get_user_milestones(self.serialized_test_user, namespace=namespace)
        self.assertEqual(len(user_milestones), 2)

        # And there should be no more Milestones required for this User+Course
        with self.assertNumQueries(2):
            required_milestones = api.get_course_required_milestones(
                self.test_course_key,
                self.serialized_test_user
            )
        self.assertEqual(len(required_milestones), 0)

 def test_get_user_milestones(self):
     """ Unit Test: test_get_user_milestones """
     with self.assertNumQueries(2):
         api.add_user_milestone(self.serialized_test_user, self.test_milestone)
     self.assertTrue(api.user_has_milestone(self.serialized_test_user, self.test_milestone))

 def test_add_user_milestone_active_exists(self):
     """ Unit Test: test_add_user_milestone """
     api.add_user_milestone(self.serialized_test_user, self.test_milestone)
     with self.assertNumQueries(1):
         api.add_user_milestone(self.serialized_test_user, self.test_milestone)
     self.assertTrue(api.user_has_milestone(self.serialized_test_user, self.test_milestone))