diff --git a/CHANGELOG.rst b/CHANGELOG.rst index e17579a..1ffefb3 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -16,6 +16,21 @@ Unreleased * +0.5.1 - 2026-03-17 +****************** + +Added +===== + +* Credential eligibility check endpoint (``GET /api/learning_credentials/v1/eligibility//``) + with detailed progress information. + +Changed +======= + +* Processor functions now return ``dict[int, dict[str, Any]]`` with detailed eligibility information instead of ``list[int]`` of eligible user IDs. +* Processor functions now accept an optional ``user_id`` parameter for single-user eligibility checks. + 0.5.0 - 2026-01-29 ****************** diff --git a/learning_credentials/api/v1/permissions.py b/learning_credentials/api/v1/permissions.py index 80955db..564658f 100644 --- a/learning_credentials/api/v1/permissions.py +++ b/learning_credentials/api/v1/permissions.py @@ -19,6 +19,23 @@ from rest_framework.views import APIView +class IsAdminOrSelf(BasePermission): + """ + Permission to allow only admins or the user themselves to access the API. + + Non-staff users cannot pass a ``username`` that is not their own. + """ + + def has_permission(self, request: "Request", view: "APIView") -> bool: # noqa: ARG002 + """Check if the user is admin or accessing their own data.""" + if request.user.is_staff: + return True + + if username := request.query_params.get("username"): + return request.user.username == username + return True + + class CanAccessLearningContext(BasePermission): """Permission to allow access to learning context if the user is enrolled.""" diff --git a/learning_credentials/api/v1/serializers.py b/learning_credentials/api/v1/serializers.py index e900165..d324309 100644 --- a/learning_credentials/api/v1/serializers.py +++ b/learning_credentials/api/v1/serializers.py @@ -1,15 +1,48 @@ """API serializers for learning credentials.""" +from typing import Any + from rest_framework import serializers from learning_credentials.models import Credential class CredentialSerializer(serializers.ModelSerializer): - """Serializer that returns credential metadata.""" + """Serializer that returns credential metadata (for the public verification endpoint).""" class Meta: """Serializer metadata.""" model = Credential fields = ('user_full_name', 'created', 'learning_context_name', 'status', 'invalidation_reason') + + +class CredentialEligibilitySerializer(serializers.Serializer): + """Serializer for credential eligibility information with dynamic fields.""" + + credential_type_id = serializers.IntegerField() + name = serializers.CharField() + is_generation_enabled = serializers.BooleanField() + is_eligible = serializers.BooleanField() + existing_credential = serializers.UUIDField(required=False, allow_null=True) + existing_credential_url = serializers.URLField(required=False, allow_blank=True, allow_null=True) + + current_grades = serializers.DictField(required=False) + required_grades = serializers.DictField(required=False) + + current_completion = serializers.FloatField(required=False, allow_null=True) + required_completion = serializers.FloatField(required=False, allow_null=True) + + steps = serializers.DictField(required=False) + + def to_representation(self, instance: dict) -> dict[str, Any]: + """Remove null/empty fields from representation.""" + data = super().to_representation(instance) + return {key: value for key, value in data.items() if value is not None and value not in ({}, [])} + + +class CredentialEligibilityResponseSerializer(serializers.Serializer): + """Serializer for the complete credential eligibility response.""" + + context_key = serializers.CharField() + credentials = CredentialEligibilitySerializer(many=True) diff --git a/learning_credentials/api/v1/urls.py b/learning_credentials/api/v1/urls.py index 7528e4e..43cabc0 100644 --- a/learning_credentials/api/v1/urls.py +++ b/learning_credentials/api/v1/urls.py @@ -2,7 +2,7 @@ from django.urls import path -from .views import CredentialConfigurationCheckView, CredentialMetadataView +from .views import CredentialConfigurationCheckView, CredentialEligibilityView, CredentialMetadataView urlpatterns = [ path( @@ -11,4 +11,9 @@ name='credential_configuration_check', ), path('metadata//', CredentialMetadataView.as_view(), name='credential-metadata'), + path( + 'eligibility//', + CredentialEligibilityView.as_view(), + name='credential-eligibility', + ), ] diff --git a/learning_credentials/api/v1/views.py b/learning_credentials/api/v1/views.py index 18fd8f6..7b3d258 100644 --- a/learning_credentials/api/v1/views.py +++ b/learning_credentials/api/v1/views.py @@ -3,6 +3,8 @@ from typing import TYPE_CHECKING import edx_api_doc_tools as apidocs +from django.contrib.auth import get_user_model +from django.shortcuts import get_object_or_404 from edx_api_doc_tools import ParameterLocation from rest_framework import status from rest_framework.permissions import IsAuthenticated @@ -11,10 +13,11 @@ from learning_credentials.models import Credential, CredentialConfiguration -from .permissions import CanAccessLearningContext -from .serializers import CredentialSerializer +from .permissions import CanAccessLearningContext, IsAdminOrSelf +from .serializers import CredentialEligibilityResponseSerializer, CredentialSerializer if TYPE_CHECKING: + from django.contrib.auth.models import User from rest_framework.request import Request @@ -141,3 +144,127 @@ def get(self, _request: "Request", uuid: str) -> Response: serializer = CredentialSerializer(credential) return Response(serializer.data, status=status.HTTP_200_OK) + + +class CredentialEligibilityView(APIView): + """ + API view for credential eligibility checking and generation. + + **GET**: Returns detailed eligibility info for all configured credentials in a learning context. + **POST**: Triggers credential generation for an eligible user. + + Staff users can operate on behalf of other users via the ``username`` parameter. + """ + + permission_classes = (IsAuthenticated, IsAdminOrSelf, CanAccessLearningContext) + + def _get_eligibility_data( + self, user: "User", config: CredentialConfiguration, credentials_by_config_id: dict[int, Credential] + ) -> dict: + """Calculate eligibility data for a credential configuration.""" + progress_data = config.get_user_eligibility_details(user_id=user.id) + existing_credential = credentials_by_config_id.get(config.id) + + return { + 'credential_type_id': config.credential_type.pk, + 'name': config.credential_type.name, + 'is_generation_enabled': config.periodic_task.enabled, + **progress_data, + 'existing_credential': existing_credential.uuid if existing_credential else None, + 'existing_credential_url': existing_credential.download_url if existing_credential else None, + } + + @apidocs.schema( + parameters=[ + apidocs.string_parameter( + "learning_context_key", + ParameterLocation.PATH, + description=( + "Learning context identifier. Can be a course key (course-v1:OpenedX+DemoX+DemoCourse) " + "or learning path key (path-v1:OpenedX+DemoX+DemoPath+Demo)" + ), + ), + apidocs.string_parameter( + "retrieval_func", + ParameterLocation.QUERY, + description=( + "Filter by credential type retrieval function " + "(e.g. learning_credentials.processors.retrieve_subsection_grades)." + ), + ), + ], + responses={ + 200: CredentialEligibilityResponseSerializer, + 400: "Invalid context key format.", + 403: "User is not authenticated.", + 404: "Learning context not found or user does not have access.", + }, + ) + def get(self, request: "Request", learning_context_key: str) -> Response: + """ + Get credential eligibility for a learning context. + + Returns detailed eligibility information for all configured credentials, including: + + - Current grades and requirements for grade-based credentials + - Completion percentages for completion-based credentials + - Step-by-step progress for learning paths + - Existing credential info if already generated + + **Query Parameters** + + - ``username`` (staff only): View eligibility for a specific user. + - ``retrieval_func``: Filter by credential type retrieval function + (e.g. ``learning_credentials.processors.retrieve_subsection_grades``). + + **Example Request** + + ``GET /api/learning_credentials/v1/eligibility/course-v1:OpenedX+DemoX+DemoCourse/`` + + **Example Response** + + .. code-block:: json + + { + "context_key": "course-v1:OpenedX+DemoX+DemoCourse", + "credentials": [ + { + "credential_type_id": 1, + "name": "Certificate of Achievement", + "is_eligible": true, + "existing_credential": null, + "current_grades": {"final exam": 86, "total": 82}, + "required_grades": {"final exam": 65, "total": 80} + } + ] + } + """ + username = request.query_params.get('username') + user = get_object_or_404(get_user_model(), username=username) if username else request.user + + configurations = CredentialConfiguration.objects.filter( + learning_context_key=learning_context_key + ).select_related('credential_type', 'periodic_task') + + retrieval_func = request.query_params.get('retrieval_func') + if retrieval_func: + configurations = configurations.filter(credential_type__retrieval_func=retrieval_func) + + # Pre-fetch all credentials for this user and learning context to avoid N+1 queries in the loop below. + credentials = Credential.objects.filter(user_id=user.id, configuration__in=configurations).exclude( + status__in=[Credential.Status.ERROR, Credential.Status.INVALIDATED] + ) + credentials_by_config_id = {credential.configuration_id: credential for credential in credentials} + + eligibility_data = [ + self._get_eligibility_data(user, config, credentials_by_config_id) for config in configurations + ] + + response_data = { + 'context_key': learning_context_key, + 'credentials': eligibility_data, + } + + serializer = CredentialEligibilityResponseSerializer(data=response_data) + serializer.is_valid(raise_exception=True) + return Response(serializer.data) diff --git a/learning_credentials/models.py b/learning_credentials/models.py index 5481d5c..55ee2e2 100644 --- a/learning_credentials/models.py +++ b/learning_credentials/models.py @@ -7,7 +7,7 @@ import uuid as uuid_lib from importlib import import_module from pathlib import Path -from typing import TYPE_CHECKING, Self +from typing import TYPE_CHECKING, Any, Self import jsonfield from django.conf import settings @@ -183,11 +183,12 @@ def filter_out_user_ids_with_credentials(self, user_ids: list[int]) -> list[int] filtered_user_ids_set = set(user_ids) - set(users_ids_with_credentials) return list(filtered_user_ids_set) - def get_eligible_user_ids(self) -> list[int]: + def _call_retrieval_func(self, user_id: int | None = None) -> dict[int, dict[str, Any]]: """ - Get the list of eligible learners for the given course. + Call the retrieval function and return detailed results. - :return: A list of user IDs. + :param user_id: Optional. If provided, only check eligibility for this user. + :return: A dict mapping user IDs to their detailed progress information. """ func_path = self.credential_type.retrieval_func module_path, func_name = func_path.rsplit('.', 1) @@ -195,7 +196,27 @@ def get_eligible_user_ids(self) -> list[int]: func = getattr(module, func_name) custom_options = _deep_merge(self.credential_type.custom_options, self.custom_options) - return func(self.learning_context_key, custom_options) + return func(self.learning_context_key, custom_options, user_id=user_id) + + def get_eligible_user_ids(self, user_id: int | None = None) -> list[int]: + """ + Get the list of eligible learners for the given learning context. + + :param user_id: Optional. If provided, only check eligibility for this user. + :return: A list of eligible user IDs. + """ + results = self._call_retrieval_func(user_id) + return [uid for uid, details in results.items() if details.get('is_eligible', False)] + + def get_user_eligibility_details(self, user_id: int) -> dict[str, Any]: + """ + Get detailed eligibility information for a specific user. + + :param user_id: The user ID to check eligibility for. + :return: Dictionary containing eligibility details and progress information. + """ + results = self._call_retrieval_func(user_id) + return results.get(user_id, {'is_eligible': False}) def generate_credential_for_user(self, user_id: int, celery_task_id: int = 0) -> Credential: """ diff --git a/learning_credentials/processors.py b/learning_credentials/processors.py index f4aeca3..027072c 100644 --- a/learning_credentials/processors.py +++ b/learning_credentials/processors.py @@ -4,6 +4,9 @@ The functions prefixed with `retrieve_` are automatically detected by the admin page and are used to retrieve the IDs of the users that meet the criteria for the credential type. +All processors return ``dict[int, dict[str, Any]]`` — a mapping from user ID to detailed progress info. +Each user's dict always includes an ``is_eligible`` boolean. + We will move this module to an external repository (a plugin). """ @@ -38,46 +41,63 @@ def _process_learning_context( learning_context_key: LearningContextKey, - course_processor: Callable[[CourseKey, dict[str, Any]], list[int]], + course_processor: Callable[[CourseKey, dict[str, Any], int | None], dict[int, dict[str, Any]]], options: dict[str, Any], -) -> list[int]: + user_id: int | None = None, +) -> dict[int, dict[str, Any]]: """ Process a learning context (course or learning path) using the given course processor function. For courses, runs the processor directly. For learning paths, runs the processor on each - course in the path with step-specific options (if available), and returns the intersection - of eligible users across all courses. + course in the path with step-specific options (if available), and returns detailed results + with per-step breakdown. Args: learning_context_key: A course key or learning path key to process - course_processor: A function that processes a single course and returns eligible user IDs + course_processor: A function that processes a single course and returns detailed progress for users options: Options to pass to the processor. For learning paths, may contain a "steps" key with step-specific options in the format: {"steps": {"": {...}}} + user_id: Optional. If provided, only process this specific user. Returns: - A list of eligible user IDs + A dict mapping user IDs to their detailed progress information. """ if learning_context_key.is_course: - return course_processor(learning_context_key, options) + return course_processor(learning_context_key, options, user_id) learning_path = LearningPath.objects.get(key=learning_context_key) - results = None - for course in learning_path.steps.all(): - course_options = options.get("steps", {}).get(str(course.course_key), options) - course_results = set(course_processor(course.course_key, course_options)) + step_results_by_course: dict[str, dict[int, dict[str, Any]]] = {} + all_user_ids: set[int] = set() + + # TODO: Use a single Completion Aggregator request when retrieving step results for a single user. + for step in learning_path.steps.all(): + course_options = options.get("steps", {}).get(str(step.course_key), options) + step_results = course_processor(step.course_key, course_options, user_id) - if results is None: - results = course_results - else: - results &= course_results + step_results_by_course[str(step.course_key)] = step_results + all_user_ids.update(step_results.keys()) # Filter out users who are not enrolled in the Learning Path. - results &= set( - learning_path.enrolled_users.filter(learningpathenrollment__is_active=True).values_list('id', flat=True), + all_user_ids &= set( + learning_path.enrolled_users.filter(learningpathenrollment__is_active=True).values_list('id', flat=True) ) - return list(results) if results else [] + final_results: dict[int, dict[str, Any]] = {} + for uid in all_user_ids: + overall_eligible = True + user_step_results: dict[str, dict[str, Any]] = {} + + for course_key, step_results in step_results_by_course.items(): + if uid in step_results: + user_step_results[course_key] = step_results[uid] + overall_eligible = overall_eligible and step_results[uid].get('is_eligible', False) + else: + overall_eligible = False + + final_results[uid] = {'is_eligible': overall_eligible, 'steps': user_step_results} + + return final_results def _get_category_weights(course_id: CourseKey) -> dict[str, float]: @@ -100,7 +120,7 @@ def _get_category_weights(course_id: CourseKey) -> dict[str, float]: return category_weight_ratios -def _get_grades_by_format(course_id: CourseKey, users: list[User]) -> dict[int, dict[str, int]]: +def _get_grades_by_format(course_id: CourseKey, users: list[User]) -> dict[int, dict[str, float]]: """ Get the grades for each user, categorized by assignment types. @@ -162,31 +182,63 @@ def _are_grades_passing_criteria( return total_score >= required_grades.get('total', 0) -def _retrieve_course_subsection_grades(course_id: CourseKey, options: dict[str, Any]) -> list[int]: - """Implementation for retrieving course grades.""" +def _calculate_grades_progress( + user_grades: dict[str, float], + required_grades: dict[str, float], + category_weights: dict[str, float], +) -> dict[str, Any]: + """ + Calculate detailed progress information for grade-based criteria. + + :param user_grades: The grades of the user, divided by category. + :param required_grades: The required grades for each category. + :param category_weights: The weight of each category. + :returns: Dict with is_eligible, current_grades, and required_grades. + """ + total_score = 0 + for category, score in user_grades.items(): + if category in category_weights: + total_score += score * category_weights[category] + + user_grades_with_total = {**user_grades, 'total': total_score} + is_eligible = _are_grades_passing_criteria(user_grades, required_grades, category_weights) + + return { + 'is_eligible': is_eligible, + 'current_grades': user_grades_with_total, + 'required_grades': required_grades, + } + + +def _retrieve_course_subsection_grades( + course_id: CourseKey, options: dict[str, Any], user_id: int | None = None +) -> dict[int, dict[str, Any]]: + """Retrieve detailed grade progress for enrolled users in a course.""" required_grades: dict[str, int] = options['required_grades'] required_grades = {key.lower(): value * 100 for key, value in required_grades.items()} - users = get_course_enrollments(course_id) + users = get_course_enrollments(course_id, user_id) grades = _get_grades_by_format(course_id, users) log.debug(grades) weights = _get_category_weights(course_id) - eligible_users = [] - for user_id, user_grades in grades.items(): - if _are_grades_passing_criteria(user_grades, required_grades, weights): - eligible_users.append(user_id) + results: dict[int, dict[str, Any]] = {} + for uid, user_grades in grades.items(): + results[uid] = _calculate_grades_progress(user_grades, required_grades, weights) - return eligible_users + return results -def retrieve_subsection_grades(learning_context_key: LearningContextKey, options: dict[str, Any]) -> list[int]: +def retrieve_subsection_grades( + learning_context_key: LearningContextKey, options: dict[str, Any], user_id: int | None = None +) -> dict[int, dict[str, Any]]: """ - Retrieve the users that have passing grades in all required categories. + Retrieve detailed grade progress for users in a learning context. :param learning_context_key: The learning context key (course or learning path). :param options: The custom options for the credential. - :returns: The IDs of the users that have passing grades in all required categories. + :param user_id: Optional. If provided, only process this specific user. + :returns: A dict mapping user IDs to their grade progress (is_eligible, current_grades, required_grades). Options: - required_grades: A dictionary of required grades for each category, where the keys are the category names and @@ -232,7 +284,7 @@ def retrieve_subsection_grades(learning_context_key: LearningContextKey, options } } """ - return _process_learning_context(learning_context_key, _retrieve_course_subsection_grades, options) + return _process_learning_context(learning_context_key, _retrieve_course_subsection_grades, options, user_id) def _prepare_request_to_completion_aggregator(course_id: CourseKey, query_params: dict, url: str) -> APIView: @@ -252,54 +304,81 @@ def _prepare_request_to_completion_aggregator(course_id: CourseKey, query_params drf_request = Request(django_request) # convert django.core.handlers.wsgi.WSGIRequest to DRF request view = CompletionDetailView() - view.request = drf_request + view.request = drf_request # type: ignore[assignment] # HACK: Bypass the API permissions. staff_user = get_user_model().objects.filter(is_staff=True).first() - view._effective_user = staff_user # noqa: SLF001 + view.request.user = staff_user log.debug('Finished preparing the request for retrieving the completion.') return view -def _retrieve_course_completions(course_id: CourseKey, options: dict[str, Any]) -> list[int]: - """Implementation for retrieving course completions.""" +def _retrieve_course_completions( + course_id: CourseKey, options: dict[str, Any], user_id: int | None = None +) -> dict[int, dict[str, Any]]: + """Retrieve detailed completion progress for enrolled users in a course.""" # If it turns out to be too slow, we can: # 1. Modify the Completion Aggregator to emit a signal/event when a user achieves a certain completion threshold. # 2. Get this data from the `Aggregator` model. Filter by `aggregation name == 'course'`, `course_key`, `percent`. required_completion = options.get('required_completion', 0.9) + # Map usernames to user IDs via enrolled users. + users = get_course_enrollments(course_id, user_id) + username_to_id = {user.username: user.id for user in users} + url = f'/completion-aggregator/v1/course/{course_id}/' # The API supports up to 10k results per page, but we limit it to 1k to avoid performance issues. query_params = {'page_size': 1000, 'page': 1} + if user_id and user_id in username_to_id.values(): + # If we are processing a single user, we can directly query their completion to get live data. + # This is because the Completion Aggregator API calculates stale completions for single user queries. + query_params['username'] = users[0].username + # TODO: Extract the logic of this view into an API. The current approach is very hacky. view = _prepare_request_to_completion_aggregator(course_id, query_params.copy(), url) - completions = [] + completions: dict[str, float] = {} while True: - # noinspection PyUnresolvedReferences response = view.get(view.request, str(course_id)) log.debug(response.data) - completions.extend( - res['username'] for res in response.data['results'] if res['completion']['percent'] >= required_completion - ) + for res in response.data['results']: + try: + completions[res['username']] = res['completion']['percent'] + except KeyError: + # If we request completion for a single user, the API does not return the username in the response. + completions[users[0].username] = res['completion']['percent'] if not response.data['pagination']['next']: break query_params['page'] += 1 view = _prepare_request_to_completion_aggregator(course_id, query_params.copy(), url) - return list(get_user_model().objects.filter(username__in=completions).values_list('id', flat=True)) + results: dict[int, dict[str, Any]] = {} + for username, current_completion in completions.items(): + if username in username_to_id: + uid = username_to_id[username] + results[uid] = { + 'is_eligible': current_completion >= required_completion, + 'current_completion': current_completion, + 'required_completion': required_completion, + } + + return results -def retrieve_completions(learning_context_key: LearningContextKey, options: dict[str, Any]) -> list[int]: +def retrieve_completions( + learning_context_key: LearningContextKey, options: dict[str, Any], user_id: int | None = None +) -> dict[int, dict[str, Any]]: """ - Retrieve the course completions for all users through the Completion Aggregator API. + Retrieve detailed completion progress for users through the Completion Aggregator API. :param learning_context_key: The learning context key (course or learning path). :param options: The custom options for the credential. - :returns: The IDs of the users that have achieved the required completion percentage. + :param user_id: Optional. If provided, only process this specific user. + :returns: A dict mapping user IDs to their completion progress + (is_eligible, current_completion, required_completion). Options: - required_completion: The minimum required completion percentage. The default value is 0.9. @@ -319,19 +398,22 @@ def retrieve_completions(learning_context_key: LearningContextKey, options: dict } } """ - return _process_learning_context(learning_context_key, _retrieve_course_completions, options) + return _process_learning_context(learning_context_key, _retrieve_course_completions, options, user_id) -def retrieve_completions_and_grades(learning_context_key: LearningContextKey, options: dict[str, Any]) -> list[int]: +def retrieve_completions_and_grades( + learning_context_key: LearningContextKey, options: dict[str, Any], user_id: int | None = None +) -> dict[int, dict[str, Any]]: """ - Retrieve the users that meet both completion and grade criteria. + Retrieve detailed progress for users that must meet both completion and grade criteria. - This processor combines the functionality of retrieve_course_completions and retrieve_subsection_grades. + This processor combines the functionality of retrieve_completions and retrieve_subsection_grades. To be eligible, learners must satisfy both sets of criteria. :param learning_context_key: The learning context key (course or learning path). :param options: The custom options for the credential. - :returns: The IDs of the users that meet both sets of criteria. + :param user_id: Optional. If provided, only process this specific user. + :returns: A dict mapping user IDs to their combined progress. Options: - required_completion: The minimum required completion percentage (default: 0.9) @@ -372,7 +454,18 @@ def retrieve_completions_and_grades(learning_context_key: LearningContextKey, op } } """ - completion_eligible_users = set(retrieve_completions(learning_context_key, options)) - grades_eligible_users = set(retrieve_subsection_grades(learning_context_key, options)) - - return list(completion_eligible_users & grades_eligible_users) + completion_results = retrieve_completions(learning_context_key, options, user_id) + grade_results = retrieve_subsection_grades(learning_context_key, options, user_id) + + # Merge results for users present in both. + combined: dict[int, dict[str, Any]] = {} + for uid in set(completion_results) & set(grade_results): + comp = completion_results[uid] + grade = grade_results[uid] + combined[uid] = { + **comp, + **grade, + 'is_eligible': comp['is_eligible'] and grade['is_eligible'], + } + + return combined diff --git a/pyproject.toml b/pyproject.toml index 1ff57f2..c626569 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "learning-credentials" -version = "0.5.0" +version = "0.5.1rc4" description = "A pluggable service for preparing Open edX credentials." dynamic = ["readme"] requires-python = ">=3.11" diff --git a/tests/conftest.py b/tests/conftest.py index 5ce58cc..ef0f13a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -102,9 +102,18 @@ def learning_path_with_courses(users: list[User]) -> LearningPath: # ============================================================================= -def _mock_retrieval_func(_context_id: LearningContextKey, _options: dict) -> list[int]: - """Mock retrieval function that returns a fixed list of user IDs.""" - return [1, 2, 3] +def _mock_retrieval_func( + _context_id: LearningContextKey, _options: dict, user_id: int | None = None +) -> dict[int, dict]: + """Mock retrieval function that returns detailed eligibility results.""" + all_results = { + 1: {'is_eligible': True}, + 2: {'is_eligible': True}, + 3: {'is_eligible': True}, + } + if user_id is not None: + return {user_id: all_results[user_id]} if user_id in all_results else {} + return all_results def _mock_generation_func(_credential: Credential, _options: dict, *, invalidate: bool = False) -> str: diff --git a/tests/test_models.py b/tests/test_models.py index b5ca0b2..f4e9256 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -134,6 +134,18 @@ def test_get_eligible_user_ids(self, mock_credential_config: CredentialConfigura eligible_user_ids = mock_credential_config.get_eligible_user_ids() assert eligible_user_ids == [1, 2, 3] + @pytest.mark.django_db + def test_get_user_eligibility_details(self, mock_credential_config: CredentialConfiguration): + """Test that get_user_eligibility_details returns details for a known user.""" + details = mock_credential_config.get_user_eligibility_details(user_id=1) + assert details == {'is_eligible': True} + + @pytest.mark.django_db + def test_get_user_eligibility_details_fallback(self, mock_credential_config: CredentialConfiguration): + """Test that get_user_eligibility_details returns default when user is not in results.""" + details = mock_credential_config.get_user_eligibility_details(user_id=999) + assert details == {'is_eligible': False} + @pytest.mark.django_db @patch('tests.conftest._mock_retrieval_func') def test_custom_options_deep_merge( diff --git a/tests/test_processors.py b/tests/test_processors.py index 39789f8..877f491 100644 --- a/tests/test_processors.py +++ b/tests/test_processors.py @@ -155,14 +155,12 @@ def test_are_grades_passing_criteria_invalid_grade_category(): @patch('learning_credentials.processors.get_course_enrollments') @patch('learning_credentials.processors._get_grades_by_format') @patch('learning_credentials.processors._get_category_weights') -@patch('learning_credentials.processors._are_grades_passing_criteria') def test_retrieve_subsection_grades( - mock_are_grades_passing_criteria: Mock, mock_get_category_weights: Mock, mock_get_grades_by_format: Mock, mock_get_course_enrollments: Mock, ): - """Test that the function returns the eligible users.""" + """Test that the function returns detailed eligibility results for users.""" course_id = Mock(spec=CourseKey) options = { 'required_grades': { @@ -173,29 +171,52 @@ def test_retrieve_subsection_grades( } users = [Mock(name="User1", id=101), Mock(name="User2", id=102)] grades = { - 101: {'homework': 0.5, 'exam': 0.9}, - 102: {'homework': 0.3, 'exam': 0.95}, + 101: {'homework': 80.0, 'exam': 95.0}, + 102: {'homework': 30.0, 'exam': 95.0}, } - required_grades = {'homework': 40.0, 'exam': 90.0, 'total': 80.0} weights = {'homework': 0.2, 'exam': 0.7, 'lab': 0.1} mock_get_course_enrollments.return_value = users mock_get_grades_by_format.return_value = grades mock_get_category_weights.return_value = weights - mock_are_grades_passing_criteria.side_effect = [True, False] result = retrieve_subsection_grades(course_id, options) - assert result == [101] - mock_get_course_enrollments.assert_called_once_with(course_id) + assert 101 in result + assert 102 in result + assert result[101]['is_eligible'] is True + assert result[102]['is_eligible'] is False + assert 'current_grades' in result[101] + assert 'required_grades' in result[101] + mock_get_course_enrollments.assert_called_once_with(course_id, None) mock_get_grades_by_format.assert_called_once_with(course_id, users) mock_get_category_weights.assert_called_once_with(course_id) - mock_are_grades_passing_criteria.assert_has_calls( - [ - call(grades[101], required_grades, weights), - call(grades[102], required_grades, weights), - ], - ) + + +@patch('learning_credentials.processors.get_course_enrollments') +@patch('learning_credentials.processors._get_grades_by_format') +@patch('learning_credentials.processors._get_category_weights') +def test_retrieve_subsection_grades_unknown_grade_category( + mock_get_category_weights: Mock, + mock_get_grades_by_format: Mock, + mock_get_course_enrollments: Mock, +): + """Test that grade categories not in the grading policy are skipped in total calculation.""" + course_id = Mock(spec=CourseKey) + + options = {'required_grades': {'exam': 0.9}} + users = [Mock(name="User1", id=101)] + grades = {101: {'homework': 80.0, 'lab': 100.0}} + weights = {'homework': 0.5, 'exam': 0.5} + + mock_get_course_enrollments.return_value = users + mock_get_grades_by_format.return_value = grades + mock_get_category_weights.return_value = weights + + result = retrieve_subsection_grades(course_id, options) + + assert result[101]['is_eligible'] is False + assert result[101]['current_grades']['total'] == 80.0 * 0.5 def test_prepare_request_to_completion_aggregator(): @@ -217,8 +238,7 @@ def test_prepare_request_to_completion_aggregator(): mock_view_class.assert_called_once() assert view.request.course_id == course_id - # noinspection PyUnresolvedReferences - assert view._effective_user is staff_user + assert view.request.user is staff_user assert isinstance(view, mock_view_class.return_value.__class__) # Create a QueryDict from the query_params dictionary. @@ -228,9 +248,11 @@ def test_prepare_request_to_completion_aggregator(): @patch('learning_credentials.processors._prepare_request_to_completion_aggregator') -@patch('learning_credentials.processors.get_user_model') -def test_retrieve_course_completions(mock_get_user_model: Mock, mock_prepare_request_to_completion_aggregator: Mock): - """Test that we retrieve the course completions for all users and return IDs of users who meet the criteria.""" +@patch('learning_credentials.processors.get_course_enrollments') +def test_retrieve_course_completions( + mock_get_course_enrollments: Mock, mock_prepare_request_to_completion_aggregator: Mock +): + """Test that we retrieve the course completions for all users and return detailed results.""" course_id = Mock(spec=CourseKey) options = {'required_completion': 0.8} completions_page1 = { @@ -244,6 +266,7 @@ def test_retrieve_course_completions(mock_get_user_model: Mock, mock_prepare_req 'results': [ {'username': 'user2', 'completion': {'percent': 0.7}}, {'username': 'user3', 'completion': {'percent': 0.8}}, + {'username': 'unenrolled_user', 'completion': {'percent': 0.95}}, ], } @@ -253,31 +276,18 @@ def test_retrieve_course_completions(mock_get_user_model: Mock, mock_prepare_req mock_view_page2.get.return_value.data = completions_page2 mock_prepare_request_to_completion_aggregator.side_effect = [mock_view_page1, mock_view_page2] - def filter_side_effect(*_args, **kwargs) -> list[int]: - """ - A mock side effect function for User.objects.filter(). - - It allows testing this code without a database access. - - :returns: The user IDs corresponding to the provided usernames. - """ - usernames = kwargs['username__in'] - - values_list_mock = Mock() - values_list_mock.return_value = [username_id_map[username] for username in usernames] - queryset_mock = Mock() - queryset_mock.values_list = values_list_mock - - return queryset_mock - - username_id_map = {"user1": 1, "user2": 2, "user3": 3} - mock_user_model = Mock() - mock_user_model.objects.filter.side_effect = filter_side_effect - mock_get_user_model.return_value = mock_user_model + # Mock enrolled users to map usernames to IDs. + mock_get_course_enrollments.return_value = [ + Mock(username='user1', id=1), + Mock(username='user2', id=2), + Mock(username='user3', id=3), + ] result = retrieve_completions(course_id, options) - assert result == [1, 3] + assert result[1] == {'is_eligible': True, 'current_completion': 0.9, 'required_completion': 0.8} + assert result[2] == {'is_eligible': False, 'current_completion': 0.7, 'required_completion': 0.8} + assert result[3] == {'is_eligible': True, 'current_completion': 0.8, 'required_completion': 0.8} mock_prepare_request_to_completion_aggregator.assert_has_calls( [ call(course_id, {'page_size': 1000, 'page': 1}, f'/completion-aggregator/v1/course/{course_id}/'), @@ -286,16 +296,57 @@ def filter_side_effect(*_args, **kwargs) -> list[int]: ) mock_view_page1.get.assert_called_once_with(mock_view_page1.request, str(course_id)) mock_view_page2.get.assert_called_once_with(mock_view_page2.request, str(course_id)) - mock_user_model.objects.filter.assert_called_once_with(username__in=['user1', 'user3']) + + +@patch('learning_credentials.processors._prepare_request_to_completion_aggregator') +@patch('learning_credentials.processors.get_course_enrollments') +def test_retrieve_course_completions_single_user( + mock_enrollments: Mock, mock_aggregator_request: Mock, course_key: CourseKey +): + """Test that single-user queries add username to params and handle missing username in response.""" + options = {'required_completion': 0.8} + user = Mock(username='user1', id=1) + mock_enrollments.return_value = [user] + + # The Completion Aggregator API omits the username for single-user queries. + completions = {'pagination': {'next': None}, 'results': [{'completion': {'percent': 0.95}}]} + mock_view = Mock() + mock_view.get.return_value.data = completions + mock_aggregator_request.return_value = mock_view + + result = retrieve_completions(course_key, options, user_id=user.id) + + assert result[user.id] == {'is_eligible': True, 'current_completion': 0.95, 'required_completion': 0.8} + mock_aggregator_request.assert_called_once_with( + course_key, + {'page_size': 1000, 'page': 1, 'username': user.username}, + f'/completion-aggregator/v1/course/{course_key}/', + ) @pytest.mark.parametrize( - ('completion_users', 'grades_users', 'expected_result'), + ('completion_results', 'grade_results', 'expected_eligible_ids'), [ - ([101, 102, 103], [102, 103, 104], [102, 103]), - ([101, 102], [103, 104], []), - ([101, 102], [101, 102], [101, 102]), - ([101, 102], [], []), + ( + {101: {'is_eligible': True}, 102: {'is_eligible': True}, 103: {'is_eligible': True}}, + {102: {'is_eligible': True}, 103: {'is_eligible': True}, 104: {'is_eligible': True}}, + {102, 103}, + ), + ( + {101: {'is_eligible': True}, 102: {'is_eligible': True}}, + {103: {'is_eligible': True}, 104: {'is_eligible': True}}, + set(), + ), + ( + {101: {'is_eligible': True}, 102: {'is_eligible': True}}, + {101: {'is_eligible': True}, 102: {'is_eligible': True}}, + {101, 102}, + ), + ( + {101: {'is_eligible': True}, 102: {'is_eligible': True}}, + {}, + set(), + ), ], ids=[ "Some users pass both criteria", @@ -309,22 +360,24 @@ def filter_side_effect(*_args, **kwargs) -> list[int]: def test_retrieve_course_completions_and_grades( mock_retrieve_completions: Mock, mock_retrieve_subsection_grades: Mock, - completion_users: list[int], - grades_users: list[int], - expected_result: list[int], + completion_results: dict, + grade_results: dict, + expected_eligible_ids: set[int], ): - """Test that the function returns the intersection of eligible users from both criteria.""" + """Test that the function merges results for users present in both criteria.""" course_id = Mock(spec=CourseKey) options = Mock() - mock_retrieve_completions.return_value = completion_users - mock_retrieve_subsection_grades.return_value = grades_users + mock_retrieve_completions.return_value = completion_results + mock_retrieve_subsection_grades.return_value = grade_results result = retrieve_completions_and_grades(course_id, options) - assert result == expected_result - mock_retrieve_completions.assert_called_once_with(course_id, options) - mock_retrieve_subsection_grades.assert_called_once_with(course_id, options) + assert set(result.keys()) == expected_eligible_ids + for uid in expected_eligible_ids: + assert result[uid]['is_eligible'] is True + mock_retrieve_completions.assert_called_once_with(course_id, options, None) + mock_retrieve_subsection_grades.assert_called_once_with(course_id, options, None) @pytest.mark.parametrize( @@ -338,28 +391,35 @@ def test_retrieve_course_completions_and_grades( @pytest.mark.django_db def test_retrieve_data_for_learning_path( patch_target: str, - function_to_test: Callable[[str, dict], list[int]], + function_to_test: Callable, learning_path_with_courses: LearningPath, users: list[User], ): - """Test retrieving data for a learning path.""" + """Test retrieving data for a learning path returns eligible users with step breakdown.""" with patch(patch_target) as mock_retrieve: options = {} - mock_retrieve.side_effect = ( - (users[i].id for i in (0, 1, 2, 4, 5)), # Users passing/completing course0 - (users[i].id for i in (0, 1, 2, 3, 4, 5)), # Users passing/completing course1 - (users[i].id for i in (0, 2, 3, 4, 5)), # Users passing/completing course2 - ) + + def make_results(user_indices: tuple) -> dict[int, dict]: + return {users[i].id: {'is_eligible': True} for i in user_indices} + + mock_retrieve.side_effect = [ + make_results((0, 1, 2, 4, 5)), # Users passing/completing course0 + make_results((0, 1, 2, 3, 4, 5)), # Users passing/completing course1 + make_results((0, 2, 3, 4, 5)), # Users passing/completing course2 + ] result = function_to_test(learning_path_with_courses.key, options) - assert sorted(result) == [users[0].id, users[2].id] + # users[0] and users[2] pass all 3 courses AND are enrolled+active in the learning path. + # users[4] passes all 3 but enrollment is inactive. users[5] is not enrolled at all. + eligible_ids = sorted(uid for uid, details in result.items() if details['is_eligible']) + assert eligible_ids == [users[0].id, users[2].id] assert mock_retrieve.call_count == 3 course_keys = [step.course_key for step in learning_path_with_courses.steps.all()] for i, course_key in enumerate(course_keys): call_args = mock_retrieve.call_args_list[i] - assert call_args[0] == (course_key, options) + assert call_args[0] == (course_key, options, None) @patch("learning_credentials.processors._retrieve_course_completions") @@ -369,6 +429,7 @@ def test_retrieve_data_for_learning_path_with_step_options( learning_path_with_courses: LearningPath, ): """Test retrieving data for a learning path with step-specific options.""" + mock_retrieve.return_value = {} course_keys = [step.course_key for step in learning_path_with_courses.steps.all()] options = { @@ -383,6 +444,6 @@ def test_retrieve_data_for_learning_path_with_step_options( retrieve_completions(learning_path_with_courses.key, options) assert mock_retrieve.call_count == 3 - assert mock_retrieve.call_args_list[0][0] == (course_keys[0], options["steps"][str(course_keys[0])]) - assert mock_retrieve.call_args_list[1][0] == (course_keys[1], options["steps"][str(course_keys[1])]) - assert mock_retrieve.call_args_list[2][0] == (course_keys[2], options) + assert mock_retrieve.call_args_list[0][0] == (course_keys[0], options["steps"][str(course_keys[0])], None) + assert mock_retrieve.call_args_list[1][0] == (course_keys[1], options["steps"][str(course_keys[1])], None) + assert mock_retrieve.call_args_list[2][0] == (course_keys[2], options, None) diff --git a/tests/test_views.py b/tests/test_views.py index dff7704..9fa33f7 100644 --- a/tests/test_views.py +++ b/tests/test_views.py @@ -12,6 +12,7 @@ from rest_framework.test import APIClient from learning_credentials.models import Credential, CredentialConfiguration +from test_utils.factories import UserFactory if TYPE_CHECKING: from django.contrib.auth.models import User @@ -261,3 +262,153 @@ def test_invalidated_credential_metadata(self, user: User, credential: Credentia assert response.status_code == status.HTTP_200_OK assert response.data['status'] == Credential.Status.INVALIDATED assert response.data['invalidation_reason'] == "Reissued due to name change." + + +@pytest.mark.django_db +class TestCredentialEligibilityView: + """Tests for the CredentialEligibilityView.""" + + def _make_get_request( + self, user: User | None, learning_context_key: LearningContextKey, **query_params + ) -> Response: + """Helper to make GET request to the eligibility endpoint.""" + client = _get_api_client(user) + url = reverse( + 'learning_credentials_api_v1:credential-eligibility', + kwargs={'learning_context_key': str(learning_context_key)}, + ) + return client.get(url, query_params) + + def test_get_unauthenticated_returns_403(self, course_key: CourseKey): + """Test that unauthenticated users get 403.""" + response = self._make_get_request(None, course_key) + assert response.status_code == status.HTTP_403_FORBIDDEN + + @patch('learning_credentials.api.v1.permissions.get_course_enrollments') + def test_get_no_configurations(self, mock_enrollments: Mock, user: User, course_key: CourseKey): + """Test that an empty credentials list is returned when no configurations exist.""" + mock_enrollments.return_value = [user] + response = self._make_get_request(user, course_key) + + assert response.status_code == status.HTTP_200_OK + assert response.data == {'context_key': str(course_key), 'credentials': []} + + @patch('learning_credentials.api.v1.permissions.get_course_enrollments') + def test_get_eligible_user( + self, mock_enrollments: Mock, user: User, course_key: CourseKey, mock_credential_config: CredentialConfiguration + ): + """Test eligibility response for an eligible user with no existing credential.""" + mock_enrollments.return_value = [user] + with patch.object(CredentialConfiguration, 'get_user_eligibility_details', return_value={'is_eligible': True}): + response = self._make_get_request(user, course_key) + + assert response.status_code == status.HTTP_200_OK + credentials = response.data['credentials'] + assert len(credentials) == 1 + cred = credentials[0] + assert cred['credential_type_id'] == mock_credential_config.credential_type.pk + assert cred['name'] == mock_credential_config.credential_type.name + assert cred['is_eligible'] is True + # None values should be stripped by serializer's to_representation. + assert 'existing_credential' not in cred + assert 'existing_credential_url' not in cred + + @patch('learning_credentials.api.v1.permissions.get_course_enrollments') + def test_get_filters_by_retrieval_func( + self, + mock_enrollments: Mock, + user: User, + course_key: CourseKey, + mock_credential_config: CredentialConfiguration, + completion_config: CredentialConfiguration, + ): + """Test that the retrieval_func query parameter filters configurations.""" + mock_enrollments.return_value = [user] + with patch.object(CredentialConfiguration, 'get_user_eligibility_details', return_value={'is_eligible': False}): + response = self._make_get_request( + user, course_key, retrieval_func=mock_credential_config.credential_type.retrieval_func + ) + + assert response.status_code == status.HTTP_200_OK + credentials = response.data['credentials'] + assert len(credentials) == 1 + assert credentials[0]['credential_type_id'] == mock_credential_config.credential_type.pk + + @patch('learning_credentials.api.v1.permissions.get_course_enrollments') + def test_get_filter_by_retrieval_func_no_match( + self, mock_enrollments: Mock, user: User, course_key: CourseKey, mock_credential_config: CredentialConfiguration + ): + """Test that filtering by a non-matching retrieval_func returns no credentials.""" + mock_enrollments.return_value = [user] + response = self._make_get_request(user, course_key, retrieval_func='nonexistent.module.func') + + assert response.status_code == status.HTTP_200_OK + assert response.data == {'context_key': str(course_key), 'credentials': []} + + @pytest.mark.usefixtures('mock_credential_config') + @patch('learning_credentials.api.v1.permissions.get_course_enrollments') + def test_get_strips_empty_dicts(self, mock_enrollments: Mock, user: User, course_key: CourseKey): + """Test that the serializer strips empty dict values from the response.""" + mock_enrollments.return_value = [user] + with patch.object( + CredentialConfiguration, + 'get_user_eligibility_details', + return_value={'is_eligible': False, 'current_grades': {}, 'steps': {}}, + ): + response = self._make_get_request(user, course_key) + + assert response.status_code == status.HTTP_200_OK + cred = response.data['credentials'][0] + assert cred['is_eligible'] is False + assert 'current_grades' not in cred + assert 'steps' not in cred + + @patch('learning_credentials.api.v1.permissions.get_course_enrollments') + def test_get_with_existing_credential( + self, mock_enrollments: Mock, user: User, course_key: CourseKey, credential: Credential + ): + """Test that existing credential info is included in the response.""" + mock_enrollments.return_value = [user] + with patch.object(CredentialConfiguration, 'get_user_eligibility_details', return_value={'is_eligible': True}): + response = self._make_get_request(user, course_key) + + assert response.status_code == status.HTTP_200_OK + cred = response.data['credentials'][0] + assert str(cred['existing_credential']) == str(credential.uuid) + assert cred['existing_credential_url'] == credential.download_url + + @patch('learning_credentials.api.v1.permissions.get_course_enrollments') + def test_get_excludes_error_and_invalidated_credentials( + self, mock_enrollments: Mock, user: User, course_key: CourseKey, mock_credential_config: CredentialConfiguration + ): + """Test that credentials with ERROR or INVALIDATED status are not returned as existing.""" + mock_enrollments.return_value = [user] + mock_credential_config.credential_set.create(user=user, status=Credential.Status.ERROR) + mock_credential_config.credential_set.create(user=user, status=Credential.Status.INVALIDATED) + with patch.object(CredentialConfiguration, 'get_user_eligibility_details', return_value={'is_eligible': True}): + response = self._make_get_request(user, course_key) + + assert response.status_code == status.HTTP_200_OK + cred = response.data['credentials'][0] + assert 'existing_credential' not in cred + + @pytest.mark.usefixtures('mock_credential_config') + def test_staff_get_for_other_user(self, staff_user: User, user: User, course_key: CourseKey): + """Test that staff can view eligibility for another user via username param.""" + with patch.object(CredentialConfiguration, 'get_user_eligibility_details', return_value={'is_eligible': False}): + response = self._make_get_request(staff_user, course_key, username=user.username) + + assert response.status_code == status.HTTP_200_OK + assert len(response.data['credentials']) == 1 + + def test_non_staff_cannot_get_for_other_user(self, user: User, course_key: CourseKey): + """Test that non-staff users cannot view eligibility for another user.""" + other_user = UserFactory() + response = self._make_get_request(user, course_key, username=other_user.username) + assert response.status_code == status.HTTP_403_FORBIDDEN + + @pytest.mark.usefixtures('mock_credential_config') + def test_get_user_not_found_returns_404(self, staff_user: User, course_key: CourseKey): + """Test that 404 is returned when the specified username doesn't exist.""" + response = self._make_get_request(staff_user, course_key, username='nonexistent_user') + assert response.status_code == status.HTTP_404_NOT_FOUND diff --git a/uv.lock b/uv.lock index 7818580..6c1950e 100644 --- a/uv.lock +++ b/uv.lock @@ -1820,7 +1820,7 @@ wheels = [ [[package]] name = "learning-credentials" -version = "0.5.0" +version = "0.5.1rc4" source = { virtual = "." } dependencies = [ { name = "celery" },