Andal.LND/edx-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add ability for notify_credentials to revoke program certs

Browse Source 
[APER-2504]

This PR adds additional functionality to the tasks kicked off when the `notify_credentials` management command is run.

I have added a new keyword arg (revoke_program_certs) that, if True, will check to see if we need to revoke a program certificate. This functionality was introduced to help automate fixing cases where a learner still has access to a Program Certificate even if they have been awarded all of the course certificates in a Program (see APER-2499).

The functionality has to be committed in two separate PRs because of our blue/green deployments. The task changes will come first, then we will update the management command to be able to set/pass the new settings. New settings were added as keyword args (defaulting to False) in order to ensure that we won't trip up our workers.
This commit is contained in:
Justin Hynes
2023-06-14 15:05:55 +00:00
parent 3fab0aec65
commit d427d404da
2 changed files with 306 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

193
openedx/core/djangoapps/credentials/tests/test_tasks.py
View File

@@ -8,15 +8,18 @@ from datetime import datetime
import ddt
import pytest
from django.conf import settings
from django.contrib.auth import get_user_model
from django.db import connection, reset_queries
from django.test import TestCase, override_settings
from freezegun import freeze_time
from opaque_keys.edx.keys import CourseKey
from common.djangoapps.student.tests.factories import UserFactory
from lms.djangoapps.certificates.api import get_recently_modified_certificates
from lms.djangoapps.certificates.data import CertificateStatuses
from lms.djangoapps.certificates.models import GeneratedCertificate
from lms.djangoapps.grades.models import PersistentCourseGrade
from lms.djangoapps.grades.models_api import get_recently_modified_grades
from lms.djangoapps.grades.tests.utils import mock_passing_grade
from lms.djangoapps.certificates.tests.factories import CertificateDateOverrideFactory, GeneratedCertificateFactory
from openedx.core.djangoapps.catalog.tests.factories import CourseFactory, CourseRunFactory, ProgramFactory
@@ -26,6 +29,8 @@ from openedx.core.djangolib.testing.utils import skip_unless_lms
from openedx.core.djangoapps.credentials.tasks.v1 import tasks
User = get_user_model()
TASKS_MODULE = 'openedx.core.djangoapps.credentials.tasks.v1.tasks'
@@ -107,8 +112,6 @@ class TestHandleNotifyCredentialsTask(TestCase):
self.cert4 = GeneratedCertificateFactory(
user=self.user2, course_id='course-v1:edX+Test+4', status=CertificateStatuses.downloadable
)
print(('self.cert1.modified_date', self.cert1.modified_date))
# No factory for these
with freeze_time(datetime(2017, 1, 1)):
self.grade1 = PersistentCourseGrade.objects.create(user_id=self.user.id, course_id='course-v1:edX+Test+1',
@@ -122,8 +125,6 @@ class TestHandleNotifyCredentialsTask(TestCase):
with freeze_time(datetime(2017, 2, 1, 5)):
self.grade4 = PersistentCourseGrade.objects.create(user_id=self.user2.id, course_id='course-v1:edX+Test+4',
percent_grade=1)
print(('self.grade1.modified', self.grade1.modified))
self.options = {
'args_from_database': False,
'auto': False,
@@ -557,3 +558,187 @@ class TestIsCourseRunInAProgramUtil(TestCase):
mock_get_programs.return_value = self.data
course_run2 = CourseRunFactory()
assert not tasks.is_course_run_in_a_program(course_run2['key'])
@ddt.ddt
@skip_unless_lms
@mock.patch('openedx.core.djangoapps.credentials.tasks.v1.tasks.handle_course_cert_changed')
@mock.patch('openedx.core.djangoapps.credentials.tasks.v1.tasks.handle_course_cert_awarded')
@mock.patch('openedx.core.djangoapps.credentials.tasks.v1.tasks.handle_course_cert_revoked')
@mock.patch('openedx.core.djangoapps.credentials.tasks.v1.tasks.send_grade_if_interesting')
class TestSendNotifications(TestCase):
"""
Unit Tests for the `send_notifications` function in the `tasks.py` file.
"""
def setUp(self):
super().setUp()
self.user = UserFactory()
self.course_run = CourseRunFactory()
self.course_key = CourseKey.from_string(self.course_run['key'])
self.course = CourseFactory(course_runs=[self.course_run])
self.site = SiteConfigurationFactory.create(
site_values={'course_org_filter': [self.course_key.org]},
)
self.certificate = GeneratedCertificateFactory(
user=self.user,
course_id=self.course_run['key'],
status=CertificateStatuses.downloadable
)
self.grade = PersistentCourseGrade.objects.create(
user_id=self.user.id,
course_id=self.course_run['key'],
percent_grade=1.0
)
def _get_certs_qs(self):
"""
The function under test expects a QuerySet containing certificate information, this retrieves the data needed
to invoke the `send_notifications` function in the form that is expected. We leverage a utility function that
the product code uses.
"""
return get_recently_modified_certificates([self.course_run['key']], None, None, [self.user.id])
def _get_grades_qs(self):
"""
The function under test expects a QuerySet containing grade information, this retrieves the data needed
to invoke the `send_notifications` function in the form that is expected. We leverage a utility function that
the product code uses.
"""
users = User.objects.filter(id__in=[self.user.id])
return get_recently_modified_grades([self.course_run['key']], None, None, users)
def _build_expected_signal_args(self):
return {
'sender': None,
'user': self.user,
'course_key': self.certificate.course_id,
'mode': self.certificate.mode,
'status': self.certificate.status,
'verbose': False
}
def test_send_notifications(self, mock_interesting, mock_revoked, mock_awarded, mock_changed):
"""
A test case that verifies the `send_notifications` functions behavior when the `notify_programs` flag is False.
"""
certs = self._get_certs_qs()
grades = self._get_grades_qs()
tasks.send_notifications(
certs,
grades,
self.site
)
expected_signal_args = self._build_expected_signal_args()
mock_changed.assert_called_once_with(**expected_signal_args)
mock_interesting.assert_called_once_with(
self.user,
self.grade.course_id,
self.certificate.mode,
self.certificate.status,
self.grade.letter_grade,
self.grade.percent_grade,
grade_last_updated=self.grade.modified,
verbose=False
)
mock_revoked.assert_not_called()
mock_awarded.assert_not_called()
@ddt.data(
['downloadable', True],
['unavailable', False]
)
@ddt.unpack
def test_send_notifications_notify_programs(
self,
cert_status,
award_expected,
mock_interesting,
mock_revoked,
mock_awarded,
mock_changed
):
"""
Test cases that verify the `send_notifications` functions behavior when the `notify_programs` flag is True.
"""
certs = self._get_certs_qs()
grades = self._get_grades_qs()
self.certificate.status = cert_status
self.certificate.save()
tasks.send_notifications(
certs,
grades,
self.site,
notify_programs=True
)
expected_signal_args = self._build_expected_signal_args()
mock_changed.assert_called_once_with(**expected_signal_args)
mock_interesting.assert_called_once_with(
self.user,
self.grade.course_id,
self.certificate.mode,
self.certificate.status,
self.grade.letter_grade,
self.grade.percent_grade,
grade_last_updated=self.grade.modified,
verbose=False
)
mock_revoked.assert_not_called()
if award_expected:
mock_awarded.assert_called_once_with(**expected_signal_args)
else:
mock_awarded.assert_not_called()
@ddt.data(
['downloadable', False],
['unavailable', True]
)
@ddt.unpack
def test_send_notifications_revoke_programs(
self,
cert_status,
revoke_expected,
mock_interesting,
mock_revoked,
mock_awarded,
mock_changed
):
"""
Test cases that verify the `send_notifications` functions behavior when the `revoke_program_certs` flag is True.
"""
certs = self._get_certs_qs()
grades = self._get_grades_qs()
self.certificate.status = cert_status
self.certificate.save()
tasks.send_notifications(
certs,
grades,
self.site,
notify_programs=True,
revoke_program_certs=True
)
expected_signal_args = self._build_expected_signal_args()
mock_changed.assert_called_once_with(**expected_signal_args)
mock_interesting.assert_called_once_with(
self.user,
self.grade.course_id,
self.certificate.mode,
self.certificate.status,
self.grade.letter_grade,
self.grade.percent_grade,
grade_last_updated=self.grade.modified,
verbose=False
)
if revoke_expected:
mock_revoked.assert_called_once_with(**expected_signal_args)
mock_awarded.mock_not_called()
else:
mock_revoked.assert_not_called()
mock_awarded.assert_called_once_with(**expected_signal_args)