From 2ba11138cc7b9f6f34d5695375f07a8c852552f1 Mon Sep 17 00:00:00 2001 From: Agrendalath Date: Wed, 25 Oct 2023 00:25:48 +0200 Subject: [PATCH] test: add tests --- openedx_certificates/compat.py | 2 - requirements/test.in | 2 + test_settings.py | 1 + test_utils/factories.py | 29 +++ tests/test_generators.py | 226 ++++++++++++++++++--- tests/test_models.py | 258 ++++++++++++++++++++++-- tests/test_processors.py | 349 +++++++++++++++++++++++++-------- 7 files changed, 746 insertions(+), 121 deletions(-) create mode 100644 test_utils/factories.py diff --git a/openedx_certificates/compat.py b/openedx_certificates/compat.py index 4e1bcf6..b367844 100644 --- a/openedx_certificates/compat.py +++ b/openedx_certificates/compat.py @@ -16,8 +16,6 @@ from django.contrib.auth.models import User from opaque_keys.edx.keys import CourseKey -# TODO: Do we still need all these pylint disable comments? We switched to ruff. - def get_celery_app() -> Celery: """Get Celery app to reuse configuration and queues.""" diff --git a/requirements/test.in b/requirements/test.in index 33ec25e..38adc3f 100644 --- a/requirements/test.in +++ b/requirements/test.in @@ -7,3 +7,5 @@ pytest-cov # pytest extension for code coverage statistics django-coverage-plugin # https://github.com/nedbat/django_coverage_plugin pytest-django # pytest extension for better Django support code-annotations # provides commands used by the pii_check make target. +dj-inmemorystorage # provides an in-memory storage backend for Django +factory-boy # provides a fixtures replacement for pytest diff --git a/test_settings.py b/test_settings.py index fe3b75a..bd617b1 100644 --- a/test_settings.py +++ b/test_settings.py @@ -70,3 +70,4 @@ def root(path: Path) -> Path: ] TESTING = True +USE_TZ = True diff --git a/test_utils/factories.py b/test_utils/factories.py new file mode 100644 index 0000000..db9a407 --- /dev/null +++ b/test_utils/factories.py @@ -0,0 +1,29 @@ +"""Factories for creating test data.""" + +from datetime import datetime + +import factory +from django.contrib.auth.models import User +from factory.django import DjangoModelFactory +from pytz import UTC + + +class UserFactory(DjangoModelFactory): + """A Factory for User objects.""" + + class Meta: # noqa: D106 + model = User + django_get_or_create = ('email', 'username') + + _DEFAULT_PASSWORD = 'test' # noqa: S105 + + username = factory.Sequence('robot{}'.format) + email = factory.Sequence('robot+test+{}@edx.org'.format) + password = factory.django.Password(_DEFAULT_PASSWORD) + first_name = factory.Sequence('Robot{}'.format) + last_name = 'Test' + is_staff = False + is_active = True + is_superuser = False + last_login = datetime(2012, 1, 1, tzinfo=UTC) + date_joined = datetime(2011, 1, 1, tzinfo=UTC) diff --git a/tests/test_generators.py b/tests/test_generators.py index 774a7dc..f52088e 100644 --- a/tests/test_generators.py +++ b/tests/test_generators.py @@ -1,34 +1,212 @@ """This module contains unit tests for the generate_pdf_certificate function.""" +from __future__ import annotations -import tempfile -import unittest -from pathlib import Path +import io +from unittest.mock import Mock, call, patch +from uuid import uuid4 -from pypdf import PdfReader +import pytest +from django.conf import settings +from django.core.files.base import ContentFile +from django.core.files.storage import DefaultStorage, FileSystemStorage +from django.test import override_settings +from inmemorystorage import InMemoryStorage +from opaque_keys.edx.keys import CourseKey +from pypdf import PdfWriter -from openedx_certificates.generators import generate_pdf_certificate +from openedx_certificates.generators import ( + _get_user_name, + _register_font, + _save_certificate, + _write_text_on_template, + generate_pdf_certificate, +) -class TestGeneratePdfCertificate(unittest.TestCase): - """Unit tests for the generate_pdf_certificate function.""" +def test_get_user_name(): + """Test the _get_user_name function.""" + user = Mock(first_name="First", last_name="Last") + user.profile.name = "Profile Name" - def test_generate_pdf_certificate(self): - """Generate a PDF certificate and check that it contains the correct data.""" - data = { - 'username': 'Test user', - 'course_name': 'Some course', - 'template_path': 'openedx_certificates/static/certificate_templates/achievement.pdf', - } + # Test when profile name is available + assert _get_user_name(user) == "Profile Name" - # Generate the PDF certificate. - with tempfile.NamedTemporaryFile(suffix='.pdf') as certificate_file: - data['output_path'] = certificate_file.name - generate_pdf_certificate(data) + # Test when profile name is not available + user.profile.name = None + assert _get_user_name(user) == "First Last" - assert Path(data['output_path']).exists() - pdf_reader = PdfReader(certificate_file) - page = pdf_reader.pages[0] - text = page.extract_text() - assert data['username'] in text - assert data['course_name'] in text +@patch("openedx_certificates.generators.ExternalCertificateAsset.get_asset_by_slug") +def test_register_font_without_custom_font(mock_get_asset_by_slug: Mock): + """Test the _register_font falls back to the default font when no custom font is specified.""" + options = {} + assert _register_font(options) == "Helvetica" + mock_get_asset_by_slug.assert_not_called() + + +@patch("openedx_certificates.generators.ExternalCertificateAsset.get_asset_by_slug") +@patch('openedx_certificates.generators.TTFont') +@patch("openedx_certificates.generators.pdfmetrics.registerFont") +def test_register_font_with_custom_font(mock_register_font: Mock, mock_font_class: Mock, mock_get_asset_by_slug: Mock): + """Test the _register_font registers the custom font when specified.""" + custom_font = "MyFont" + options = {"font": custom_font} + + mock_get_asset_by_slug.return_value = "font_path" + + assert _register_font(options) == custom_font + mock_get_asset_by_slug.assert_called_once_with(custom_font) + mock_font_class.assert_called_once_with(custom_font, mock_get_asset_by_slug.return_value) + mock_register_font.assert_called_once_with(mock_font_class.return_value) + + +@pytest.mark.parametrize( + ("username", "course_name", "options"), + [ + ('John Doe', 'Programming 101', {}), # No options - use default coordinates. + ('John Doe', 'Programming 101', {'name_y': 250, 'course_name_y': 200}), # Custom coordinates. + ], +) +@patch('openedx_certificates.generators.canvas.Canvas', return_value=Mock(stringWidth=Mock(return_value=10))) +def test_write_text_on_template(mock_canvas_class: Mock, username: str, course_name: str, options: dict[str, int]): + """Test the _write_text_on_template function.""" + template_height = 300 + template_width = 200 + font = 'Helvetica' + string_width = mock_canvas_class.return_value.stringWidth.return_value + + # Reset the mock to discard calls list from previous tests + mock_canvas_class.reset_mock() + + template_mock = Mock() + template_mock.mediabox = [0, 0, template_width, template_height] + + # Call the function with test parameters and mocks + _write_text_on_template(template_mock, font, username, course_name, options) + + # Verifying that Canvas was the correct pagesize. + # Use `call_args_list` to ignore the first argument, which is an instance of io.BytesIO. + assert mock_canvas_class.call_args_list[0][1]['pagesize'] == (template_width, template_height) + + # Mock Canvas object retrieved from Canvas constructor call + canvas_object = mock_canvas_class.return_value + + # Expected coordinates for drawString method, based on fixed stringWidth + expected_name_x = (template_width - string_width) / 2 + expected_name_y = options.get('name_y', 290) + expected_course_name_x = (template_width - string_width) / 2 + expected_course_name_y = options.get('course_name_y', 220) + + # Check the calls to setFont and drawString methods on Canvas object + assert canvas_object.setFont.call_args_list[0] == call(font, 32) + assert canvas_object.drawString.call_args_list[0] == call(expected_name_x, expected_name_y, username) + + assert canvas_object.setFont.call_args_list[1] == call(font, 28) + assert canvas_object.drawString.call_args_list[1] == call( + expected_course_name_x, expected_course_name_y, course_name + ) + + +@override_settings(LMS_ROOT_URL="http://example.com", MEDIA_URL="media/") +@pytest.mark.parametrize( + "storage", + [ + (InMemoryStorage()), # Test a real storage, without mocking. + (Mock(spec=FileSystemStorage, exists=Mock(return_value=False))), # Test calls in a mocked storage. + # Test calls in a mocked storage when the file already exists. + (Mock(spec=FileSystemStorage, exists=Mock(return_value=True))), + ], +) +@patch('openedx_certificates.generators.ContentFile', autospec=True) +def test_save_certificate(mock_contentfile: Mock, storage: DefaultStorage | Mock): + """Test the _save_certificate function.""" + # Mock the certificate. + certificate = Mock(spec=PdfWriter) + certificate_uuid = uuid4() + output_path = f'external_certificates/{certificate_uuid}.pdf' + pdf_bytes = io.BytesIO() + certificate.write.return_value = pdf_bytes + content_file = ContentFile(pdf_bytes.getvalue()) + mock_contentfile.return_value = content_file + + # Run the function. + with patch('openedx_certificates.generators.default_storage', storage): + url = _save_certificate(certificate, certificate_uuid) + + # Check the calls in a mocked storage. + if isinstance(storage, Mock): + storage.exists.assert_called_once_with(output_path) + storage.save.assert_called_once_with(output_path, content_file) + storage.url.assert_not_called() + if storage.exists.return_value: + storage.delete.assert_called_once_with(output_path) + else: + storage.delete.assert_not_called() + + if isinstance(storage, Mock): + assert url == f'{settings.LMS_ROOT_URL}/media/{output_path}' + else: + assert url == f'/{output_path}' + + +@pytest.mark.parametrize( + ("course_name", "options", "expected_template_slug"), + [ + ('Test Course', {'template': 'template_slug'}, 'template_slug'), + ('Test Course;Test Course', {'template': 'template_slug'}, 'template_slug'), + ( + 'Test Course;Test Course', + {'template': 'template_slug', 'template_two-lines': 'template_two_lines_slug'}, + 'template_two_lines_slug', + ), + ], +) +@patch( + 'openedx_certificates.generators.ExternalCertificateAsset.get_asset_by_slug', + return_value=Mock( + open=Mock( + return_value=Mock( + __enter__=Mock(return_value=Mock(read=Mock(return_value=b'pdf_data'))), __exit__=Mock(return_value=None) + ) + ) + ), +) +@patch('openedx_certificates.generators._get_user_name') +@patch('openedx_certificates.generators.get_course_name') +@patch('openedx_certificates.generators._register_font') +@patch('openedx_certificates.generators.PdfReader') +@patch('openedx_certificates.generators.PdfWriter') +@patch( + 'openedx_certificates.generators._write_text_on_template', + return_value=Mock(getpdfdata=Mock(return_value=b'pdf_data')), +) +@patch('openedx_certificates.generators._save_certificate', return_value='certificate_url') +def test_generate_pdf_certificate( # noqa: PLR0913 + mock_save_certificate: Mock, + mock_write_text_on_template: Mock, + mock_pdf_writer: Mock, + mock_pdf_reader: Mock, + mock_register_font: Mock, + mock_get_course_name: Mock, + mock_get_user_name: Mock, + mock_get_asset_by_slug: Mock, + course_name: str, + options: dict[str, str], + expected_template_slug: str, +): + """Test the generate_pdf_certificate function.""" + course_id = CourseKey.from_string('course-v1:edX+DemoX+Demo_Course') + user = Mock() + mock_get_course_name.return_value = course_name + + result = generate_pdf_certificate(course_id, user, Mock(), options) + + assert result == 'certificate_url' + mock_get_asset_by_slug.assert_called_with(expected_template_slug) + mock_get_user_name.assert_called_once_with(user) + mock_get_course_name.assert_called_once_with(course_id) + mock_register_font.assert_called_once_with(options) + mock_pdf_reader.assert_called() + mock_pdf_writer.assert_called() + mock_write_text_on_template.assert_called_once() + mock_save_certificate.assert_called_once() diff --git a/tests/test_models.py b/tests/test_models.py index 7209834..3fcc785 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -1,21 +1,249 @@ -"""Tests for the `openedx-certificates` generators module.""" +"""Tests for the `openedx-certificates` models.""" +from __future__ import annotations -import os -import tempfile +from typing import TYPE_CHECKING, Any +from unittest.mock import Mock, patch +from uuid import UUID, uuid4 -from openedx_certificates.generators import generate_pdf_certificate +import pytest +from django.core.exceptions import ValidationError +from django.db import IntegrityError +from openedx_certificates.exceptions import CertificateGenerationError +from openedx_certificates.models import ( + ExternalCertificate, + ExternalCertificateCourseConfiguration, + ExternalCertificateType, +) +from test_utils.factories import UserFactory -class TestGeneratePdfCertificate: - """Tests for the `generate_pdf_certificate` function.""" +if TYPE_CHECKING: + from django.contrib.auth.models import User + from opaque_keys.edx.keys import CourseKey - def test_generate_pdf_certificate(self): - """Test that the function generates a PDF certificate.""" - data = { - 'username': 'Test user', - 'course_name': 'Some course', - 'template_path': 'openedx_certificates/static/certificate_templates/achievement.pdf', - 'output_path': tempfile.NamedTemporaryFile(suffix='.pdf').name, + +def _mock_retrieval_func(_course_id: CourseKey, _options: dict[str, Any]) -> list[int]: + return [1, 2, 3] + + +def _mock_generation_func(_course_id: CourseKey, _user: User, _certificate_uuid: UUID, _options: dict[str, Any]) -> str: + return "test_url" + + +class TestExternalCertificateType: + """Tests for the ExternalCertificateType model.""" + + def test_str(self): + """Test the string representation of the model.""" + certificate_type = ExternalCertificateType(name="Test Type") + assert str(certificate_type) == "Test Type" + + def test_clean_with_valid_functions(self): + """Test the clean method with valid function paths.""" + certificate_type = ExternalCertificateType( + name="Test Type", + retrieval_func="test_models._mock_retrieval_func", + generation_func="test_models._mock_generation_func", + ) + certificate_type.clean() + + @pytest.mark.parametrize("function_path", ["", "invalid_format_func"]) + def test_clean_with_invalid_function_format(self, function_path: str): + """Test the clean method with invalid function format.""" + certificate_type = ExternalCertificateType( + name="Test Type", + retrieval_func="test_models._mock_retrieval_func", + generation_func=function_path, + ) + with pytest.raises(ValidationError) as exc: + certificate_type.clean() + assert "Function path must be in format 'module.function_name'" in str(exc.value) + + def test_clean_with_invalid_function(self): + """Test the clean method with invalid function paths.""" + certificate_type = ExternalCertificateType( + name="Test Type", + retrieval_func="test_models._mock_retrieval_func", + generation_func="invalid.module.path", + ) + with pytest.raises(ValidationError) as exc: + certificate_type.clean() + assert ( + f"The function {certificate_type.generation_func} could not be found. Please provide a valid path" + in str(exc.value) + ) + + +class TestExternalCertificateCourseConfiguration: + """Tests for the ExternalCertificateCourseConfiguration model.""" + + def setup_method(self): + """Prepare the test data.""" + self.certificate_type = ExternalCertificateType( + name="Test Type", + retrieval_func="test_models._mock_retrieval_func", + generation_func="test_models._mock_generation_func", + ) + self.course_config = ExternalCertificateCourseConfiguration( + course_id="course-v1:TestX+T101+2023", certificate_type=self.certificate_type + ) + + @pytest.mark.django_db() + def test_periodic_task_is_auto_created(self): + """Test that a periodic task is automatically created for the new configuration.""" + self.certificate_type.save() + self.course_config.save() + self.course_config.refresh_from_db() + + assert (periodic_task := self.course_config.periodic_task) is not None + assert periodic_task.enabled is False + assert periodic_task.name == str(self.course_config) + assert periodic_task.args == f'[{self.course_config.id}]' + + def test_str_representation(self): + """Test the string representation of the model.""" + assert str(self.course_config) == f'{self.certificate_type.name} in course-v1:TestX+T101+2023' + + def test_get_eligible_user_ids(self): + """Test the get_eligible_user_ids method.""" + eligible_user_ids = self.course_config.get_eligible_user_ids() + assert eligible_user_ids == [1, 2, 3] + + @pytest.mark.xfail(reason="The filtering is currently disabled for testing purposes.") + @pytest.mark.django_db() + def test_filter_out_user_ids_with_certificates(self): + """Test the filter_out_user_ids_with_certificates method.""" + self.certificate_type.save() + self.course_config.save() + + cert_data = { + "course_id": self.course_config.course_id, + "certificate_type": self.certificate_type.name, + } + + ExternalCertificate.objects.create( + uuid=uuid4(), user_id=1, status=ExternalCertificate.Status.GENERATING, **cert_data + ) + ExternalCertificate.objects.create( + uuid=uuid4(), user_id=2, status=ExternalCertificate.Status.AVAILABLE, **cert_data + ) + ExternalCertificate.objects.create( + uuid=uuid4(), user_id=3, status=ExternalCertificate.Status.ERROR, **cert_data + ) + ExternalCertificate.objects.create( + uuid=uuid4(), user_id=4, status=ExternalCertificate.Status.INVALIDATED, **cert_data + ) + ExternalCertificate.objects.create( + uuid=uuid4(), user_id=5, status=ExternalCertificate.Status.ERROR, **cert_data + ) + + filtered_users = self.course_config.filter_out_user_ids_with_certificates([1, 2, 3, 4, 6]) + assert filtered_users == [3, 6] + + @pytest.mark.django_db() + def test_generate_certificate_for_user(self): + """Test the generate_certificate_for_user method.""" + user = UserFactory.create() + task_id = 123 + + self.course_config.generate_certificate_for_user(user.id, task_id) + assert ExternalCertificate.objects.filter( + user_id=user.id, + course_id=self.course_config.course_id, + certificate_type=self.certificate_type, + user_full_name=f"{user.first_name} {user.last_name}", + status=ExternalCertificate.Status.AVAILABLE, + generation_task_id=task_id, + download_url="test_url", + ).exists() + + @pytest.mark.django_db() + def test_generate_certificate_for_user_update_existing(self): + """Test the generate_certificate_for_user method updates an existing certificate.""" + user = UserFactory.create() + + ExternalCertificate.objects.create( + user_id=user.id, + course_id=self.course_config.course_id, + certificate_type=self.certificate_type, + user_full_name="Random Name", + status=ExternalCertificate.Status.ERROR, + generation_task_id=123, + download_url="random_url", + ) + + self.course_config.generate_certificate_for_user(user.id) + assert ExternalCertificate.objects.filter( + user_id=user.id, + course_id=self.course_config.course_id, + certificate_type=self.certificate_type, + user_full_name=f"{user.first_name} {user.last_name}", + status=ExternalCertificate.Status.AVAILABLE, + generation_task_id=0, + download_url="test_url", + ).exists() + + @pytest.mark.django_db() + @patch('openedx_certificates.models.import_module') + def test_generate_certificate_for_user_with_exception(self, mock_module: Mock): + """Test the generate_certificate_for_user handles the case when the generation function raises an exception.""" + user = UserFactory.create() + task_id = 123 + + def mock_func_raise_exception(*_args, **_kwargs): + raise RuntimeError('Test Exception') + + mock_module.return_value = mock_func_raise_exception + + # Call the method under test and check that it raises the correct exception. + with pytest.raises(CertificateGenerationError) as exc: + self.course_config.generate_certificate_for_user(user.id, task_id) + + assert 'Failed to generate the' in str(exc.value) + assert ExternalCertificate.objects.filter( + user_id=user.id, + course_id=self.course_config.course_id, + certificate_type=self.certificate_type, + user_full_name=f"{user.first_name} {user.last_name}", + status=ExternalCertificate.Status.ERROR, + generation_task_id=task_id, + download_url='', + ).exists() + + +class TestExternalCertificate: + """Tests for the ExternalCertificate model.""" + + def setup_method(self): + """Prepare the test data.""" + self.certificate = ExternalCertificate( + uuid=uuid4(), + user_id=1, + user_full_name='Test User', + course_id='course-v1:TestX+T101+2023', + certificate_type='Test Type', + status=ExternalCertificate.Status.GENERATING, + download_url='http://www.test.com', + generation_task_id='12345', + ) + + def test_str_representation(self): + """Test the string representation of a certificate.""" + assert str(self.certificate) == 'Test Type for Test User in course-v1:TestX+T101+2023' + + @pytest.mark.django_db() + def test_unique_together_constraint(self): + """Test that the unique_together constraint is enforced.""" + self.certificate.save() + certificate_2_info = { + "uuid": uuid4(), + "user_id": 1, + "user_full_name": 'Test User 2', + "course_id": 'course-v1:TestX+T101+2023', + "certificate_type": 'Test Type', + "status": ExternalCertificate.Status.GENERATING, + "download_url": 'http://www.test2.com', + "generation_task_id": '122345', } - generate_pdf_certificate(data) - assert os.path.isfile(data['output_path']) + with pytest.raises(IntegrityError): + ExternalCertificate.objects.create(**certificate_2_info) diff --git a/tests/test_processors.py b/tests/test_processors.py index 2db151c..a125c0a 100644 --- a/tests/test_processors.py +++ b/tests/test_processors.py @@ -1,91 +1,280 @@ -"""This module contains unit tests for the processors module.""" +"""Tests for the certificate processors.""" +from __future__ import annotations -import unittest -from unittest.mock import Mock, patch +from unittest.mock import Mock, call, patch import pytest +from django.http import QueryDict from opaque_keys.edx.keys import CourseKey -from openedx_certificates.processors import User, retrieve_course_completions, retrieve_subsection_grades - - -class TestProcessors(unittest.TestCase): - """Unit tests for the processors module.""" - - @pytest.mark.django_db() - @patch('openedx_certificates.processors.get_section_grades_breakdown_view') - @patch('openedx_certificates.processors.get_course_grading_policy') - def test_retrieve_subsection_grades(self, mock_policy: Mock, mock_view: Mock): - """Test that retrieve_subsection_grades returns the expected results.""" - User.objects.create(username='ecommerce_worker', is_staff=True, is_superuser=True) - - course_id = CourseKey.from_string('org/course/run') - user_id = 1 - - # Mock the response from the API view. - mock_response = { - 'results': [ - { - 'username': 'user1', - 'section_breakdown': [ - {'category': 'Category 1', 'detail': 'Detail 1', 'percent': 50}, - {'category': 'Category 1', 'detail': 'Detail 2', 'percent': 50}, - {'category': 'Category 2', 'detail': 'Detail 3', 'percent': 75}, - {'category': 'Category 2', 'detail': 'Detail 4', 'percent': 25}, - {'category': 'Exam', 'detail': 'Detail 5', 'percent': 80}, - ], - }, - { - 'username': 'user2', - 'section_breakdown': [ - {'category': 'Category 1', 'detail': 'Detail 1', 'percent': 100}, - {'category': 'Category 2', 'detail': 'Detail 2', 'percent': 50}, - {'category': 'Exam', 'detail': 'Detail 3', 'percent': 70}, - ], - }, - ], - } - mock_view.return_value.get.return_value.data = mock_response +# noinspection PyProtectedMember +from openedx_certificates.processors import ( + _are_grades_passing_criteria, + _get_category_weights, + _get_grades_by_format, + _prepare_request_to_completion_aggregator, + retrieve_course_completions, + retrieve_subsection_grades, +) + + +@patch( + 'openedx_certificates.processors.get_course_grading_policy', + return_value=[{'type': 'Homework', 'weight': 0.15}, {'type': 'Exam', 'weight': 0.85}], +) +def test_get_category_weights(mock_get_course_grading_policy: Mock): + """Check that the course grading policy is retrieved and the category weights are calculated correctly.""" + course_id = Mock(spec=CourseKey) + assert _get_category_weights(course_id) == {'homework': 0.15, 'exam': 0.85} + mock_get_course_grading_policy.assert_called_once_with(course_id) + + +@patch('openedx_certificates.processors.prefetch_course_grades') +@patch('openedx_certificates.processors.get_course_grade_factory') +def test_get_grades_by_format(mock_get_course_grade_factory: Mock, mock_prefetch_course_grades: Mock): + """Test that grades are retrieved for each user and categorized by assignment types.""" + course_id = Mock(spec=CourseKey) + users = [Mock(name="User1", id=101), Mock(name="User2", id=102)] - # Mock the grading policy. - mock_policy.return_value = [ - {'type': 'Category 1', 'weight': 0.2}, - {'type': 'Category 2', 'weight': 0.3}, - {'type': 'Exam', 'weight': 0.5}, + mock_read_grades = Mock() + mock_read_grades.return_value.graded_subsections_by_format.return_value = { + 'Homework': {'subsection1': Mock(graded_total=Mock(earned=50.0, possible=100.0))}, + 'Exam': {'subsection2': Mock(graded_total=Mock(earned=90.0, possible=100.0))}, + } + mock_get_course_grade_factory.return_value.read = mock_read_grades + + result = _get_grades_by_format(course_id, users) + + assert result == {101: {'homework': 50.0, 'exam': 90.0}, 102: {'homework': 50.0, 'exam': 90.0}} + mock_prefetch_course_grades.assert_called_once_with(course_id, users) + mock_get_course_grade_factory.assert_called_once() + + mock_read_grades.assert_has_calls( + [ + call(users[0], course_key=course_id), + call().graded_subsections_by_format(), + call(users[1], course_key=course_id), + call().graded_subsections_by_format(), ] + ) - # TODO: Fix this. - # This function should return a boolean if the user passed the threshold. - # Call the function and check the results. - expected_results = { - 'user1': 0.6, - 'user2': 0.65, - } - results = retrieve_subsection_grades(course_id, user_id) - assert results == expected_results - - @pytest.mark.django_db() - @patch('openedx_certificates.processors.CompletionDetailView') - def test_retrieve_course_completions(self, mock_view): - """Test that retrieve_course_completions returns the expected results.""" - User.objects.create(username='ecommerce_worker', is_staff=True, is_superuser=True) - - course_id = CourseKey.from_string('org/course/run') - required_completion = 0.5 - - # Mock the response from the API view. - mock_response = { - 'results': [ - {'username': 'user1', 'completion': {'percent': 0.4}}, - {'username': 'user2', 'completion': {'percent': 0.6}}, - {'username': 'user3', 'completion': {'percent': 0.7}}, - ], - 'pagination': {'next': None}, +_are_grades_passing_criteria_test_data = [ + ( + "All grades are passing", + {"homework": 90, "lab": 90, "exam": 90}, + {"homework": 85, "lab": 80, "exam": 60, "total": 50}, + {"homework": 0.3, "lab": 0.3, "exam": 0.4}, + True, + ), + ( + "The homework grade is failing", + {"homework": 80, "lab": 90, "exam": 70}, + {"homework": 85, "lab": 80, "exam": 60, "total": 50}, + {"homework": 0.3, "lab": 0.3, "exam": 0.4}, + False, + ), + ( + "The total grade is failing", + {"homework": 90, "lab": 90, "exam": 70}, + {"homework": 85, "lab": 80, "exam": 60, "total": 300}, + {"homework": 0.3, "lab": 0.3, "exam": 0.4}, + False, + ), + ( + "Only the total grade is required", + {"homework": 90, "lab": 90, "exam": 70}, + {"total": 50}, + {"homework": 0.3, "lab": 0.3, "exam": 0.4}, + True, + ), + ( + "Total grade is not required", + {"homework": 90, "lab": 90, "exam": 70}, + {"homework": 85, "lab": 80}, + {"homework": 0.3, "lab": 0.3, "exam": 0.4}, + True, + ), + ( + "Required grades are not defined", + {"homework": 80, "lab": 90, "exam": 70}, + {}, + {"homework": 0.3, "lab": 0.3, "exam": 0.4}, + True, + ), + ( + "User has no grades", + {}, + {"homework": 85, "lab": 80, "exam": 60, "total": 240}, + {"homework": 0.3, "lab": 0.3, "exam": 0.4}, + False, + ), + ("User has no grades and the required grades are not defined", {}, {}, {}, True), + ( + "User has no grades in a required category", + {"homework": 90, "lab": 85}, + {"homework": 85, "lab": 80, "exam": 60}, + {"homework": 0.3, "lab": 0.3, "exam": 0.4}, + False, + ), +] + + +@pytest.mark.parametrize( + ('desc', 'user_grades', 'required_grades', 'category_weights', 'expected'), + _are_grades_passing_criteria_test_data, + ids=[i[0] for i in _are_grades_passing_criteria_test_data], +) +def test_are_grades_passing_criteria( + desc: str, # noqa: ARG001 + user_grades: dict[str, float], + required_grades: dict[str, float], + category_weights: dict[str, float], + expected: bool, +): + """Test that the user grades are compared to the required grades correctly.""" + assert _are_grades_passing_criteria(user_grades, required_grades, category_weights) == expected + + +def test_are_grades_passing_criteria_invalid_grade_category(): + """Test that an exception is raised if user grades contain a category that is not defined in the grading policy.""" + with pytest.raises(ValueError, match='unknown_category'): + _are_grades_passing_criteria( + {"homework": 90, "unknown_category": 90}, + {"total": 175}, + {"homework": 0.5, "lab": 0.5}, + ) + + +@patch('openedx_certificates.processors.get_course_enrollments') +@patch('openedx_certificates.processors._get_grades_by_format') +@patch('openedx_certificates.processors._get_category_weights') +@patch('openedx_certificates.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.""" + course_id = Mock(spec=CourseKey) + options = { + 'required_grades': { + 'homework': 0.4, + 'exam': 0.9, + 'total': 0.8, } - mock_view.return_value.get.return_value.data = mock_response + } + 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}, + } + 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) + 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), + ] + ) + + +def test_prepare_request_to_completion_aggregator(): + """Test that the request to the completion aggregator API is prepared correctly.""" + course_id = Mock(spec=CourseKey) + query_params = {'param1': 'value1', 'param2': 'value2'} + url = '/test_url/' + + with patch('openedx_certificates.processors.get_user_model') as mock_get_user_model, patch( + 'openedx_certificates.processors.CompletionDetailView' + ) as mock_view_class: + staff_user = Mock(is_staff=True) + mock_get_user_model().objects.filter().first.return_value = staff_user + + view = _prepare_request_to_completion_aggregator(course_id, query_params, url) + + mock_view_class.assert_called_once() + assert view.request.course_id == course_id + # noinspection PyUnresolvedReferences + assert view._effective_user is staff_user + assert isinstance(view, mock_view_class.return_value.__class__) + + # Create a QueryDict from the query_params dictionary. + query_params_qdict = QueryDict('', mutable=True) + query_params_qdict.update(query_params) + assert view.request.query_params.urlencode() == query_params_qdict.urlencode() + + +@patch('openedx_certificates.processors._prepare_request_to_completion_aggregator') +@patch('openedx_certificates.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.""" + course_id = Mock(spec=CourseKey) + options = {'required_completion': 0.8} + completions_page1 = { + 'pagination': {'next': '/completion-aggregator/v1/course/{course_id}/?page=2&page_size=1000'}, + 'results': [ + {'username': 'user1', 'completion': {'percent': 0.9}}, + ], + } + completions_page2 = { + 'pagination': {'next': None}, + 'results': [ + {'username': 'user2', 'completion': {'percent': 0.7}}, + {'username': 'user3', 'completion': {'percent': 0.8}}, + ], + } - # Call the function and check the results. - expected_results = ['user2', 'user3'] - results = retrieve_course_completions(course_id, required_completion) - assert results == expected_results + mock_view_page1 = Mock() + mock_view_page1.get.return_value.data = completions_page1 + mock_view_page2 = Mock() + 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 + + result = retrieve_course_completions(course_id, options) + + assert result == [1, 3] + mock_prepare_request_to_completion_aggregator.assert_has_calls( + [ + call(course_id, {'page_size': 1000, 'page': 1}, f'/completion-aggregator/v1/course/{course_id}/'), + call(course_id, {'page_size': 1000, 'page': 2}, f'/completion-aggregator/v1/course/{course_id}/'), + ] + ) + 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'])