151bd13666
* Generate common/djangoapps import shims for LMS * Generate common/djangoapps import shims for Studio * Stop appending project root to sys.path * Stop appending common/djangoapps to sys.path * Import from common.djangoapps.course_action_state instead of course_action_state * Import from common.djangoapps.course_modes instead of course_modes * Import from common.djangoapps.database_fixups instead of database_fixups * Import from common.djangoapps.edxmako instead of edxmako * Import from common.djangoapps.entitlements instead of entitlements * Import from common.djangoapps.pipline_mako instead of pipeline_mako * Import from common.djangoapps.static_replace instead of static_replace * Import from common.djangoapps.student instead of student * Import from common.djangoapps.terrain instead of terrain * Import from common.djangoapps.third_party_auth instead of third_party_auth * Import from common.djangoapps.track instead of track * Import from common.djangoapps.util instead of util * Import from common.djangoapps.xblock_django instead of xblock_django * Add empty common/djangoapps/__init__.py to fix pytest collection * Fix pylint formatting violations * Exclude import_shims/ directory tree from linting
105 lines
3.1 KiB
Python
105 lines
3.1 KiB
Python
"""
|
|
Test entitlements tasks
|
|
"""
|
|
|
|
|
|
from datetime import datetime, timedelta
|
|
|
|
import mock
|
|
import pytz
|
|
from django.test import TestCase
|
|
|
|
from common.djangoapps.entitlements import tasks
|
|
from common.djangoapps.entitlements.models import CourseEntitlementPolicy
|
|
from common.djangoapps.entitlements.tests.factories import CourseEntitlementFactory
|
|
from openedx.core.djangolib.testing.utils import skip_unless_lms
|
|
|
|
|
|
def make_entitlement(expired=False):
|
|
age = CourseEntitlementPolicy.DEFAULT_EXPIRATION_PERIOD_DAYS
|
|
past_datetime = datetime.now(tz=pytz.UTC) - timedelta(days=age)
|
|
expired_at = past_datetime if expired else None
|
|
entitlement = CourseEntitlementFactory.create(created=past_datetime, expired_at=expired_at)
|
|
return entitlement
|
|
|
|
|
|
def boom():
|
|
raise Exception('boom')
|
|
|
|
|
|
@skip_unless_lms
|
|
class TestExpireOldEntitlementsTask(TestCase):
|
|
"""
|
|
Tests for the 'expire_old_entitlements' method.
|
|
"""
|
|
def test_checks_expiration(self):
|
|
"""
|
|
Test that we actually do check expiration on each entitlement (happy path)
|
|
"""
|
|
make_entitlement()
|
|
make_entitlement()
|
|
|
|
with mock.patch(
|
|
'common.djangoapps.entitlements.models.CourseEntitlement.expired_at_datetime',
|
|
new_callable=mock.PropertyMock
|
|
) as mock_datetime:
|
|
tasks.expire_old_entitlements.delay(1, 3).get()
|
|
|
|
self.assertEqual(mock_datetime.call_count, 2)
|
|
|
|
def test_only_unexpired(self):
|
|
"""
|
|
Verify that only unexpired entitlements are included
|
|
"""
|
|
# Create an old expired and an old unexpired entitlement
|
|
make_entitlement(expired=True)
|
|
make_entitlement()
|
|
|
|
with mock.patch(
|
|
'common.djangoapps.entitlements.models.CourseEntitlement.expired_at_datetime',
|
|
new_callable=mock.PropertyMock
|
|
) as mock_datetime:
|
|
tasks.expire_old_entitlements.delay(1, 3).get()
|
|
|
|
# Make sure only the unexpired one gets used
|
|
self.assertEqual(mock_datetime.call_count, 1)
|
|
|
|
def test_retry(self):
|
|
"""
|
|
Test that we retry when an exception occurs while checking old
|
|
entitlements.
|
|
"""
|
|
make_entitlement()
|
|
|
|
with mock.patch(
|
|
'common.djangoapps.entitlements.models.CourseEntitlement.expired_at_datetime',
|
|
new_callable=mock.PropertyMock,
|
|
side_effect=boom
|
|
) as mock_datetime:
|
|
task = tasks.expire_old_entitlements.delay(1, 2)
|
|
|
|
self.assertRaises(Exception, task.get)
|
|
self.assertEqual(mock_datetime.call_count, tasks.MAX_RETRIES + 1)
|
|
|
|
|
|
@skip_unless_lms
|
|
class TestExpireOldEntitlementsTaskIntegration(TestCase):
|
|
"""
|
|
Tests for the 'expire_old_entitlements' method without mocking.
|
|
"""
|
|
def test_actually_expired(self):
|
|
"""
|
|
Integration test with CourseEntitlement to make sure we are calling the
|
|
correct API.
|
|
"""
|
|
entitlement = make_entitlement()
|
|
|
|
# Sanity check
|
|
self.assertIsNone(entitlement.expired_at)
|
|
|
|
# Run enforcement
|
|
tasks.expire_old_entitlements.delay(1, 2).get()
|
|
entitlement.refresh_from_db()
|
|
|
|
self.assertIsNotNone(entitlement.expired_at)
|