edx-platform/lms/djangoapps/instructor_analytics/tests/test_basic.py

"""
Tests for instructor.basic
"""


from unittest.mock import MagicMock, Mock, patch

import datetime
import random
import ddt
import json  # lint-amnesty, pylint: disable=wrong-import-order
from django.contrib.auth import get_user_model
from edx_proctoring.api import create_exam
from edx_proctoring.models import ProctoredExamStudentAttempt
from opaque_keys.edx.locator import UsageKey
from lms.djangoapps.instructor_analytics.basic import (  # lint-amnesty, pylint: disable=unused-import
    AVAILABLE_FEATURES,
    ENROLLMENT_FEATURES,
    PROFILE_FEATURES,
    PROGRAM_ENROLLMENT_FEATURES,
    STUDENT_FEATURES,
    StudentModule,
    issued_certificates,
    enrolled_students_features,
    get_available_features,
    get_proctored_exam_results,
    get_response_state,
    get_student_features_with_custom,
    list_may_enroll,
    list_problem_responses
)
from lms.djangoapps.program_enrollments.tests.factories import ProgramEnrollmentFactory
from lms.djangoapps.certificates.api import create_generated_certificate
from openedx.core.djangoapps.course_groups.tests.helpers import CohortFactory
from openedx.core.djangoapps.site_configuration.tests.factories import SiteConfigurationFactory
from common.djangoapps.student.models import CourseEnrollment, CourseEnrollmentAllowed
from common.djangoapps.student.tests.factories import InstructorFactory
from common.djangoapps.student.tests.factories import UserFactory
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase  # lint-amnesty, pylint: disable=wrong-import-order
from xmodule.modulestore.tests.factories import CourseFactory  # lint-amnesty, pylint: disable=wrong-import-order

User = get_user_model()


@ddt.ddt
class TestAnalyticsBasic(ModuleStoreTestCase):
    """ Test basic analytics functions. """

    def setUp(self):
        super().setUp()
        self.course_key = self.store.make_course_key('robot', 'course', 'id')
        self.users = tuple(UserFactory() for _ in range(30))
        self.ces = tuple(CourseEnrollment.enroll(user, self.course_key)
                         for user in self.users)
        self.instructor = InstructorFactory(course_key=self.course_key)
        for user in self.users:
            user.profile.meta = json.dumps({
                "position": f"edX expert {user.id}",
                "company": f"Open edX Inc {user.id}",
            })
            user.profile.save()
        self.students_who_may_enroll = list(self.users) + [UserFactory() for _ in range(5)]
        for student in self.students_who_may_enroll:
            CourseEnrollmentAllowed.objects.create(
                email=student.email,
                course_id=self.course_key,
                user=student if student in self.users else None,
            )

    @ddt.data(
        ('あなた', 'スの中'),
        ("ГЂіи lіиэ ъэтшээи", "Ђэаvэи аиↁ Ђэѓэ")
    )
    @ddt.unpack
    def test_get_response_state_with_ora(self, files_descriptions, saved_response):
        """
        Tests that ORA response state is transformed expectedly when the problem
        state contains unicode characters.
        """
        payload_state = json.dumps({
            'saved_response': json.dumps({'parts': [{'text': saved_response}]}),
            'saved_files_descriptions': json.dumps([files_descriptions]),
        })
        response = Mock(module_type='openassessment', student=Mock(username='staff'), state=payload_state)

        transformed_state = json.loads(get_response_state(response))
        assert transformed_state['saved_files_descriptions'][0] == files_descriptions
        assert transformed_state['saved_response']['parts'][0]['text'] == saved_response

    def test_list_problem_responses(self):
        def result_factory(result_id):
            """
            Return a dummy StudentModule object that can be queried for
            relevant info (student.username and state).
            """
            result = Mock(spec=['student', 'state'])
            result.student.username.return_value = f'user{result_id}'
            result.state.return_value = f'state{result_id}'
            return result

        # Ensure that UsageKey.from_string returns a problem key that list_problem_responses can work with
        # (even when called with a dummy location):
        mock_problem_key = Mock(return_value='')
        mock_problem_key.course_key = self.course_key
        with patch.object(UsageKey, 'from_string') as patched_from_string:
            patched_from_string.return_value = mock_problem_key

            # Ensure that StudentModule.objects.filter returns a result set that list_problem_responses can work with
            # (this keeps us from having to create fixtures for this test):
            mock_results = MagicMock(return_value=[result_factory(n) for n in range(5)])
            with patch.object(StudentModule, 'objects') as patched_manager:
                patched_manager.filter.return_value = mock_results

                mock_problem_location = ''
                problem_responses = list_problem_responses(self.course_key, problem_location=mock_problem_location)

                # Check if list_problem_responses called UsageKey.from_string to look up problem key:
                patched_from_string.assert_called_once_with(mock_problem_location)
                # Check if list_problem_responses called StudentModule.objects.filter to obtain relevant records:
                patched_manager.filter.assert_called_once_with(
                    course_id=self.course_key, module_state_key=mock_problem_key
                )

                # Check if list_problem_responses returned expected results:
                assert len(problem_responses) == len(mock_results)
                for mock_result in mock_results:
                    assert {'username': mock_result.student.username, 'state': mock_result.state} in problem_responses

    def test_enrolled_students_features_username(self):
        assert 'username' in AVAILABLE_FEATURES
        userreports = enrolled_students_features(self.course_key, ['username'])
        assert len(userreports) == len(self.users)
        for userreport in userreports:
            assert list(userreport.keys()) == ['username']
            assert userreport['username'] in [user.username for user in self.users]

    def test_issued_certificates_basic(self):
        """
        Test that the `issued_certificates` function returns the correct aggregated data
        for a single downloadable certificate.

        This test:
        - Creates a downloadable certificate for a user.
        - Verifies that the function returns a list with a single item.
        - Confirms that the returned certificate contains the expected course ID.
        - Ensures that the total count of issued certificates is correct.
        - Verifies that the 'report_run_date' field is present in the result.

        The test ensures that the `issued_certificates` function behaves as expected
        for a single downloadable certificate scenario.
        """
        cert_args = {
            "user": UserFactory(),
            "course_id": self.course_key,
            "mode": "honor",
            "status": "downloadable",
            "grade": "Pass"
        }
        create_generated_certificate(cert_args)
        features = ['course_id', 'mode', 'status', 'grade']
        results = issued_certificates(self.course_key, features)

        assert isinstance(results, list)
        assert len(results) == 1

        cert = results[0]
        assert cert['course_id'] == str(self.course_key)
        assert cert['total_issued_certificate'] == 1
        assert 'report_run_date' in cert

    def test_enrolled_students_features_keys(self):
        query_features = ('username', 'name', 'email', 'city', 'country',)
        for user in self.users:
            user.profile.city = f"Mos Eisley {user.id}"
            user.profile.country = f"Tatooine {user.id}"
            user.profile.save()
        for feature in query_features:
            assert feature in AVAILABLE_FEATURES
        with self.assertNumQueries(2):
            userreports = enrolled_students_features(self.course_key, query_features)
        assert len(userreports) == len(self.users)

        userreports = sorted(userreports, key=lambda u: u["username"])
        users = sorted(self.users, key=lambda u: u.username)
        for userreport, user in zip(userreports, users):
            assert set(userreport.keys()) == set(query_features)
            assert userreport['username'] == user.username
            assert userreport['email'] == user.email
            assert userreport['name'] == user.profile.name
            assert userreport['city'] == user.profile.city
            assert userreport['country'] == user.profile.country

    def test_enrolled_student_with_no_country_city(self):
        userreports = enrolled_students_features(self.course_key, ('username', 'city', 'country',))
        for userreport in userreports:
            # This behaviour is somewhat inconsistent: None string fields
            # objects are converted to "None", but non-JSON serializable fields
            # are converted to an empty string.
            assert userreport['city'] == 'None'
            assert userreport['country'] == ''

    def test_enrolled_students_meta_features_keys(self):
        """
        Assert that we can query individual fields in the 'meta' field in the UserProfile
        """
        query_features = ('meta.position', 'meta.company')
        with self.assertNumQueries(2):
            userreports = enrolled_students_features(self.course_key, query_features)
        assert len(userreports) == len(self.users)
        for userreport in userreports:
            assert set(userreport.keys()) == set(query_features)
            assert userreport['meta.position'] in [f"edX expert {user.id}" for user in self.users]
            assert userreport['meta.company'] in [f"Open edX Inc {user.id}" for user in self.users]

    def test_enrolled_students_enrollment_verification(self):
        """
        Assert that we can get enrollment mode and verification status
        """
        query_features = ('enrollment_mode', 'verification_status')
        userreports = enrolled_students_features(self.course_key, query_features)
        assert len(userreports) == len(self.users)
        # by default all users should have "audit" as their enrollment mode
        # and "N/A" as their verification status
        for userreport in userreports:
            assert set(userreport.keys()) == set(query_features)
            assert userreport['enrollment_mode'] in ['audit']
            assert userreport['verification_status'] in ['N/A']
        # make sure that the user report respects whatever value
        # is returned by verification and enrollment code
        MODE_MTHD_NAME = "common.djangoapps.student.models.course_enrollment.CourseEnrollment.enrollment_mode_for_user"
        with patch(MODE_MTHD_NAME) as enrollment_patch:
            with patch(
                "lms.djangoapps.verify_student.services.IDVerificationService.verification_status_for_user"
            ) as verify_patch:
                enrollment_patch.return_value = ["verified"]
                verify_patch.return_value = "dummy verification status"
                userreports = enrolled_students_features(self.course_key, query_features)
                assert len(userreports) == len(self.users)
                for userreport in userreports:
                    assert set(userreport.keys()) == set(query_features)
                    assert userreport['enrollment_mode'] in ['verified']
                    assert userreport['verification_status'] in ['dummy verification status']

    def test_enrolled_students_features_keys_cohorted(self):
        course = CourseFactory.create(org="test", course="course1", display_name="run1")
        course.cohort_config = {'cohorted': True, 'auto_cohort': True, 'auto_cohort_groups': ['cohort']}
        self.store.update_item(course, self.instructor.id)
        cohorted_students = [UserFactory.create() for _ in range(10)]
        cohort = CohortFactory.create(name='cohort', course_id=course.id, users=cohorted_students)
        cohorted_usernames = [student.username for student in cohorted_students]
        non_cohorted_student = UserFactory.create()
        for student in cohorted_students:
            cohort.users.add(student)
            CourseEnrollment.enroll(student, course.id)
        CourseEnrollment.enroll(non_cohorted_student, course.id)
        instructor = InstructorFactory(course_key=course.id)
        self.client.login(username=instructor.username, password='test')

        query_features = ('username', 'cohort')
        # There should be a constant of 3 SQL queries when calling
        # enrolled_students_features. The first query comes from the call to
        # CourseEnrollment.objects.filter(...) with select_related for user and profile,
        # the second comes from prefetch_related('user__course_groups'), and the third
        # comes from configuration_helpers.get_value_for_org() for custom student attributes.
        with self.assertNumQueries(3):
            userreports = enrolled_students_features(course.id, query_features)
        assert len([r for r in userreports if r['username'] in cohorted_usernames]) == len(cohorted_students)
        assert len([r for r in userreports if r['username'] == non_cohorted_student.username]) == 1
        for report in userreports:
            assert set(report.keys()) == set(query_features)
            if report['username'] in cohorted_usernames:
                assert report['cohort'] == cohort.name
            else:
                assert report['cohort'] == '[unassigned]'

    def test_enrolled_student_features_external_user_keys(self):
        query_features = ('username', 'name', 'email', 'city', 'country', 'external_user_key')
        username_with_external_user_key_dict = {}
        for i in range(len(self.users)):
            # Setup some users with ProgramEnrollments
            if i % 2 == 0:
                user = self.users[i]
                external_user_key = f'{user.username}_{i}'
                ProgramEnrollmentFactory.create(user=user, external_user_key=external_user_key)
                username_with_external_user_key_dict[user.username] = external_user_key

        with self.assertNumQueries(3):
            userreports = enrolled_students_features(self.course_key, query_features)
        assert len(userreports) == 30
        for report in userreports:
            username = report['username']
            external_key = username_with_external_user_key_dict.get(username)
            if external_key:
                assert external_key == report['external_user_key']
            else:
                assert '' == report['external_user_key']

    def test_available_features(self):
        assert len(AVAILABLE_FEATURES) == len(
            STUDENT_FEATURES +
            PROFILE_FEATURES +
            PROGRAM_ENROLLMENT_FEATURES +
            ENROLLMENT_FEATURES
        )
        assert set(AVAILABLE_FEATURES) == set(
            STUDENT_FEATURES +
            PROFILE_FEATURES +
            PROGRAM_ENROLLMENT_FEATURES +
            ENROLLMENT_FEATURES
        )

    def test_enrolled_students_enrollment_date(self):
        """Test that enrollment_date feature works correctly and returns the correct enrollment date."""
        query_features = ('username', 'enrollment_date',)
        for feature in query_features:
            assert feature in AVAILABLE_FEATURES
        with self.assertNumQueries(2):
            userreports = enrolled_students_features(self.course_key, query_features)
        assert len(userreports) == len(self.users)

        enrollment_dict = {}
        for user in self.users:
            enrollment = CourseEnrollment.objects.get(user=user, course_id=self.course_key)
            enrollment_dict[user.username] = enrollment.created

        for userreport in userreports:
            expected_enrollment_date = enrollment_dict[userreport['username']]
            assert userreport['enrollment_date'] == expected_enrollment_date

    def test_enrolled_students_extended_model_age(self):
        """Test that custom age attribute works correctly with user profile year_of_birth."""
        SiteConfigurationFactory.create(
            site_values={
                'course_org_filter': ['robot'],
                'additional_student_profile_attributes': ['age'],
            }
        )

        def get_age(self):
            return datetime.datetime.now().year - self.profile.year_of_birth
        User.age = property(get_age)

        for user in self.users:
            user.profile.year_of_birth = random.randint(1900, 2000)
            user.profile.save()

        query_features = ('username', 'age',)
        with self.assertNumQueries(3):
            userreports = enrolled_students_features(self.course_key, query_features)
        assert len(userreports) == len(self.users)

        userreports = sorted(userreports, key=lambda u: u["username"])
        users = sorted(self.users, key=lambda u: u.username)
        for userreport, user in zip(userreports, users):
            assert set(userreport.keys()) == set(query_features)
            assert userreport['age'] == str(user.age)

        del User.age

    def test_list_may_enroll(self):
        may_enroll = list_may_enroll(self.course_key, ['email'])
        assert len(may_enroll) == (len(self.students_who_may_enroll) - len(self.users))
        email_adresses = [student.email for student in self.students_who_may_enroll]
        for student in may_enroll:
            assert list(student.keys()) == ['email']
            assert student['email'] in email_adresses

    def test_get_student_exam_attempt_features(self):
        query_features = [
            'email',
            'exam_name',
            'allowed_time_limit_mins',
            'is_sample_attempt',
            'started_at',
            'completed_at',
            'status',
            'Suspicious Count',
            'Suspicious Comments',
            'Rules Violation Count',
            'Rules Violation Comments',
            'track'
        ]

        proctored_exam_id = create_exam(self.course_key, 'Test Content', 'Test Exam', 1)
        ProctoredExamStudentAttempt.create_exam_attempt(
            proctored_exam_id, self.users[0].id,
            'Test Code 1', True, False, 'ad13'
        )
        ProctoredExamStudentAttempt.create_exam_attempt(
            proctored_exam_id, self.users[1].id,
            'Test Code 2', True, False, 'ad13'
        )
        ProctoredExamStudentAttempt.create_exam_attempt(
            proctored_exam_id, self.users[2].id,
            'Test Code 3', True, False, 'asd'
        )

        proctored_exam_attempts = get_proctored_exam_results(self.course_key, query_features)
        assert len(proctored_exam_attempts) == 3
        for proctored_exam_attempt in proctored_exam_attempts:
            assert set(proctored_exam_attempt.keys()) == set(query_features)

    def test_get_student_features_with_custom_attributes(self):
        """Test that get_student_features_with_custom works with custom attributes."""

        # Test without custom attributes - should return standard features
        features = get_student_features_with_custom(self.course_key)
        assert features == STUDENT_FEATURES

        # Test with custom attributes
        SiteConfigurationFactory.create(
            site_values={
                'course_org_filter': ['robot'],
                'additional_student_profile_attributes': ['employee_id', 'department'],
            }
        )

        features = get_student_features_with_custom(self.course_key)
        expected = STUDENT_FEATURES + ('employee_id', 'department')
        assert features == expected

    def test_enrolled_students_multiple_custom_fields(self):
        """Test that multiple custom fields work correctly together."""
        SiteConfigurationFactory.create(
            site_values={
                'course_org_filter': ['robot'],
                'additional_student_profile_attributes': [
                    'student_id',
                    'employment_status',
                    'graduation_year'
                ],
            }
        )

        def get_student_id(self):
            """Generate dummy student ID"""
            try:
                return f"ID{self.id:05d}"
            except AttributeError:
                return None

        def get_employment_status(self):
            """Generate dummy employment status"""
            try:
                statuses = ['Student', 'Employed', 'Unemployed', 'Self-employed', 'Retired']
                return statuses[self.id % len(statuses)]
            except AttributeError:
                return None

        def get_graduation_year(self):
            """Generate dummy graduation year"""
            try:
                return str(2020 + (self.id % 10))
            except AttributeError:
                return None

        User.student_id = property(get_student_id)
        User.employment_status = property(get_employment_status)
        User.graduation_year = property(get_graduation_year)

        query_features = ('username', 'student_id', 'employment_status', 'graduation_year')
        with self.assertNumQueries(3):
            userreports = enrolled_students_features(self.course_key, query_features)

        assert len(userreports) == len(self.users)

        for userreport in userreports:
            assert set(userreport.keys()) == set(query_features)
            # Verify all custom fields have values
            assert userreport['student_id'] is not None
            assert userreport['student_id'].startswith('ID')
            assert userreport['employment_status'] in ['Student', 'Employed', 'Unemployed', 'Self-employed', 'Retired']
            assert userreport['graduation_year'] in [str(year) for year in range(2020, 2030)]

        del User.student_id
        del User.employment_status
        del User.graduation_year

    def get_badge_count(self):
        """Generate dummy badge count"""
        try:
            return str(self.id % 10)  # 0-9 badges
        except AttributeError:
            return "0"

    def test_custom_attributes_without_org_filter(self):
        """Test that custom attributes require course_org_filter to work properly."""
        # Create configuration without course_org_filter
        SiteConfigurationFactory.create(
            site_values={
                'additional_student_profile_attributes': ['badge_count'],
            }
        )

        User.badge_count = property(self.get_badge_count)

        # Without org filter, custom attributes should NOT be added
        features = get_student_features_with_custom(self.course_key)
        # Should return only standard features (no badge_count)
        assert features == STUDENT_FEATURES

        # Clean up
        del User.badge_count

    def test_custom_attributes_with_non_matching_org_filter(self):
        """Test that custom attributes don't work with non-matching course_org_filter."""
        # Create configuration with course_org_filter that DOESN'T match our test course org
        SiteConfigurationFactory.create(
            site_values={
                'course_org_filter': ['different_org'],
                'additional_student_profile_attributes': ['badge_count'],
            }
        )

        # With non-matching org filter, custom attributes should NOT be added
        features = get_student_features_with_custom(self.course_key)
        # Should return only standard features (no badge_count)
        assert features == STUDENT_FEATURES

    def test_get_available_features_includes_additional_attributes(self):
        """
        get_available_features should include additional_student_profile_attributes
        for the org, on top of the standard features.
        """
        SiteConfigurationFactory.create(
            site_values={
                'course_org_filter': ['robot'],
                'additional_student_profile_attributes': ['employee_id', 'department'],
            }
        )

        features = get_available_features(self.course_key)

        # Decompose what we expect:
        # student part = STUDENT_FEATURES + additional
        expected_student = STUDENT_FEATURES + ('employee_id', 'department')
        expected_all = (
            expected_student
            + PROFILE_FEATURES
            + PROGRAM_ENROLLMENT_FEATURES
            + ENROLLMENT_FEATURES
        )

        assert features == expected_all