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

Add retirement partner reporting queue and APIs

Browse Source
This commit is contained in:
bmedx
2018-06-05 14:20:57 -04:00
parent 6600e8b7e6
commit 8fbe12e4cc
8 changed files with 1700 additions and 1226 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
openedx/core/djangoapps/user_api/accounts/serializers.py
View File

@@ -14,7 +14,11 @@ from six import text_type
from lms.djangoapps.badges.utils import badges_enabled
from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
from openedx.core.djangoapps.user_api import errors
from openedx.core.djangoapps.user_api.models import RetirementState, UserRetirementStatus, UserPreference
from openedx.core.djangoapps.user_api.models import (
RetirementState,
UserPreference,
UserRetirementStatus
)
from openedx.core.djangoapps.user_api.serializers import ReadOnlyFieldsSerializerMixin
from student.models import UserProfile, LanguageProficiency, SocialLink
@@ -209,7 +213,9 @@ class AccountLegacyProfileSerializer(serializers.HyperlinkedModelSerializer, Rea
return new_name
def validate_language_proficiencies(self, value):
""" Enforce all languages are unique. """
"""
Enforce all languages are unique.
"""
language_proficiencies = [language for language in value]
unique_language_proficiencies = set(language["code"] for language in language_proficiencies)
if len(language_proficiencies) != len(unique_language_proficiencies):
@@ -217,7 +223,9 @@ class AccountLegacyProfileSerializer(serializers.HyperlinkedModelSerializer, Rea
return value
def validate_social_links(self, value):
""" Enforce only one entry for a particular social platform. """
"""
Enforce only one entry for a particular social platform.
"""
social_links = [social_link for social_link in value]
unique_social_links = set(social_link["platform"] for social_link in social_links)
if len(social_links) != len(unique_social_links):
@@ -225,29 +233,41 @@ class AccountLegacyProfileSerializer(serializers.HyperlinkedModelSerializer, Rea
return value
def transform_gender(self, user_profile, value): # pylint: disable=unused-argument
""" Converts empty string to None, to indicate not set. Replaced by to_representation in version 3. """
"""
Converts empty string to None, to indicate not set. Replaced by to_representation in version 3.
"""
return AccountLegacyProfileSerializer.convert_empty_to_None(value)
def transform_country(self, user_profile, value): # pylint: disable=unused-argument
""" Converts empty string to None, to indicate not set. Replaced by to_representation in version 3. """
"""
Converts empty string to None, to indicate not set. Replaced by to_representation in version 3.
"""
return AccountLegacyProfileSerializer.convert_empty_to_None(value)
def transform_level_of_education(self, user_profile, value): # pylint: disable=unused-argument
""" Converts empty string to None, to indicate not set. Replaced by to_representation in version 3. """
"""
Converts empty string to None, to indicate not set. Replaced by to_representation in version 3.
"""
return AccountLegacyProfileSerializer.convert_empty_to_None(value)
def transform_bio(self, user_profile, value): # pylint: disable=unused-argument
""" Converts empty string to None, to indicate not set. Replaced by to_representation in version 3. """
"""
Converts empty string to None, to indicate not set. Replaced by to_representation in version 3.
"""
return AccountLegacyProfileSerializer.convert_empty_to_None(value)
@staticmethod
def convert_empty_to_None(value):
""" Helper method to convert empty string to None (other values pass through). """
"""
Helper method to convert empty string to None (other values pass through).
"""
return None if value == "" else value
@staticmethod
def get_profile_image(user_profile, user, request=None):
""" Returns metadata about a user's profile image. """
"""
Returns metadata about a user's profile image.
"""
data = {'has_image': user_profile.has_profile_image}
urls = get_profile_image_urls_for_user(user, request)
data.update({
@@ -257,7 +277,9 @@ class AccountLegacyProfileSerializer(serializers.HyperlinkedModelSerializer, Rea
return data
def get_requires_parental_consent(self, user_profile):
""" Returns a boolean representing whether the user requires parental controls. """
"""
Returns a boolean representing whether the user requires parental controls.
"""
return user_profile.requires_parental_consent()
def _get_profile_image(self, user_profile):
@@ -374,8 +396,27 @@ class UserRetirementStatusSerializer(serializers.ModelSerializer):
exclude = ['responses', ]
class UserRetirementPartnerReportSerializer(serializers.Serializer):
"""
Perform serialization for the UserRetirementPartnerReportingStatus model
"""
original_username = serializers.CharField()
original_email = serializers.EmailField()
original_name = serializers.CharField()
orgs = serializers.ListField(child=serializers.CharField())
# Required overrides of abstract base class methods, but we don't use them
def create(self, validated_data):
pass
def update(self, instance, validated_data):
pass
def get_extended_profile(user_profile):
"""Returns the extended user profile fields stored in user_profile.meta"""
"""
Returns the extended user profile fields stored in user_profile.meta
"""
# pick the keys from the site configuration
extended_profile_field_names = configuration_helpers.get_value('extended_profile_fields', [])
@@ -395,7 +436,9 @@ def get_extended_profile(user_profile):
def get_profile_visibility(user_profile, user, configuration=None):
"""Returns the visibility level for the specified user profile."""
"""
Returns the visibility level for the specified user profile.
"""
if user_profile.requires_parental_consent():
return PRIVATE_VISIBILITY

1478
openedx/core/djangoapps/user_api/accounts/tests/test_retirement_views.py Normal file
View File

@@ -0,0 +1,1478 @@
# -*- coding: utf-8 -*-
"""
Test cases to cover account retirement views
"""
from __future__ import print_function
import datetime
import json
import unittest
from consent.models import DataSharingConsent
from django.conf import settings
from django.contrib.auth.models import User
from django.contrib.sites.models import Site
from django.core import mail
from django.core.cache import cache
from django.core.urlresolvers import reverse
from django.test import TestCase
from enterprise.models import (
EnterpriseCustomer,
EnterpriseCustomerUser,
EnterpriseCourseEnrollment,
PendingEnterpriseCustomerUser,
)
from integrated_channels.sap_success_factors.models import (
SapSuccessFactorsLearnerDataTransmissionAudit
)
import mock
from opaque_keys.edx.keys import CourseKey
import pytz
from rest_framework import status
from six import text_type
from social_django.models import UserSocialAuth
from wiki.models import ArticleRevision, Article
from wiki.models.pluginbase import RevisionPluginRevision, RevisionPlugin
from xmodule.modulestore.tests.factories import CourseFactory
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from entitlements.models import CourseEntitlementSupportDetail
from entitlements.tests.factories import CourseEntitlementFactory
from lms.djangoapps.verify_student.tests.factories import SoftwareSecurePhotoVerificationFactory
from openedx.core.djangoapps.api_admin.models import ApiAccessRequest
from openedx.core.djangoapps.credit.models import (
CreditRequirementStatus, CreditRequest, CreditCourse, CreditProvider, CreditRequirement
)
from openedx.core.djangoapps.course_groups.models import CourseUserGroup, UnregisteredLearnerCohortAssignments
from openedx.core.djangoapps.site_configuration.tests.factories import SiteFactory
from openedx.core.djangoapps.user_api.accounts.signals import USER_RETIRE_MAILINGS
from openedx.core.djangoapps.user_api.models import (
RetirementState,
UserRetirementStatus,
UserRetirementPartnerReportingStatus,
UserOrgTag
)
from openedx.core.lib.token_utils import JwtBuilder
from student.models import (
CourseEnrollment,
CourseEnrollmentAllowed,
ManualEnrollmentAudit,
PasswordHistory,
PendingEmailChange,
PendingNameChange,
Registration,
SocialLink,
UserProfile,
get_retired_username_by_username,
get_retired_email_by_email,
)
from student.tests.factories import (
ContentTypeFactory,
CourseEnrollmentAllowedFactory,
PendingEmailChangeFactory,
PermissionFactory,
SuperuserFactory,
UserFactory
)
from survey.models import SurveyAnswer
from ..views import AccountRetirementView, USER_PROFILE_PII
from ...tests.factories import UserOrgTagFactory
def build_jwt_headers(user):
"""
Helper function for creating headers for the JWT authentication.
"""
token = JwtBuilder(user).build_token([])
headers = {
'HTTP_AUTHORIZATION': 'JWT ' + token
}
return headers
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestAccountDeactivation(TestCase):
"""
Tests the account deactivation endpoint.
"""
def setUp(self):
super(TestAccountDeactivation, self).setUp()
self.test_user = UserFactory()
self.url = reverse('accounts_deactivation', kwargs={'username': self.test_user.username})
def assert_activation_status(self, headers, expected_status=status.HTTP_200_OK, expected_activation_status=False):
"""
Helper function for making a request to the deactivation endpoint, and asserting the status.
Args:
expected_status(int): Expected request's response status.
expected_activation_status(bool): Expected user has_usable_password attribute value.
"""
self.assertTrue(self.test_user.has_usable_password())
response = self.client.post(self.url, **headers)
self.assertEqual(response.status_code, expected_status)
self.test_user.refresh_from_db()
self.assertEqual(self.test_user.has_usable_password(), expected_activation_status)
def test_superuser_deactivates_user(self):
"""
Verify a user is deactivated when a superuser posts to the deactivation endpoint.
"""
superuser = SuperuserFactory()
headers = build_jwt_headers(superuser)
self.assert_activation_status(headers)
def test_user_with_permission_deactivates_user(self):
"""
Verify a user is deactivated when a user with permission posts to the deactivation endpoint.
"""
user = UserFactory()
permission = PermissionFactory(
codename='can_deactivate_users',
content_type=ContentTypeFactory(
app_label='student'
)
)
user.user_permissions.add(permission)
headers = build_jwt_headers(user)
self.assertTrue(self.test_user.has_usable_password())
self.assert_activation_status(headers)
def test_unauthorized_rejection(self):
"""
Verify unauthorized users cannot deactivate accounts.
"""
headers = build_jwt_headers(self.test_user)
self.assert_activation_status(
headers,
expected_status=status.HTTP_403_FORBIDDEN,
expected_activation_status=True
)
def test_on_jwt_headers_rejection(self):
"""
Verify users who are not JWT authenticated are rejected.
"""
UserFactory()
self.assert_activation_status(
{},
expected_status=status.HTTP_401_UNAUTHORIZED,
expected_activation_status=True
)
class RetirementTestCase(TestCase):
"""
Test case with a helper methods for retirement
"""
@classmethod
def setUpClass(cls):
super(RetirementTestCase, cls).setUpClass()
cls.setup_states()
@staticmethod
def setup_states():
"""
Create basic states that mimic our current understanding of the retirement process
"""
default_states = [
('PENDING', 1, False, True),
('LOCKING_ACCOUNT', 20, False, False),
('LOCKING_COMPLETE', 30, False, False),
('RETIRING_CREDENTIALS', 40, False, False),
('CREDENTIALS_COMPLETE', 50, False, False),
('RETIRING_ECOM', 60, False, False),
('ECOM_COMPLETE', 70, False, False),
('RETIRING_FORUMS', 80, False, False),
('FORUMS_COMPLETE', 90, False, False),
('RETIRING_EMAIL_LISTS', 100, False, False),
('EMAIL_LISTS_COMPLETE', 110, False, False),
('RETIRING_ENROLLMENTS', 120, False, False),
('ENROLLMENTS_COMPLETE', 130, False, False),
('RETIRING_NOTES', 140, False, False),
('NOTES_COMPLETE', 150, False, False),
('NOTIFYING_PARTNERS', 160, False, False),
('PARTNERS_NOTIFIED', 170, False, False),
('RETIRING_LMS', 180, False, False),
('LMS_COMPLETE', 190, False, False),
('ERRORED', 200, True, True),
('ABORTED', 210, True, True),
('COMPLETE', 220, True, True),
]
for name, ex, dead, req in default_states:
RetirementState.objects.create(
state_name=name,
state_execution_order=ex,
is_dead_end_state=dead,
required=req
)
def _create_retirement(self, state, create_datetime=None):
"""
Helper method to create a RetirementStatus with useful defaults
"""
if create_datetime is None:
create_datetime = datetime.datetime.now(pytz.UTC) - datetime.timedelta(days=8)
user = UserFactory()
return UserRetirementStatus.objects.create(
user=user,
original_username=user.username,
original_email=user.email,
original_name=user.profile.name,
retired_username=get_retired_username_by_username(user.username),
retired_email=get_retired_email_by_email(user.email),
current_state=state,
last_state=state,
responses="",
created=create_datetime,
modified=create_datetime
)
def _retirement_to_dict(self, retirement, all_fields=False):
"""
Return a dict format of this model to a consistent format for serialization, removing the long text field
`responses` for performance reasons.
"""
retirement_dict = {
u'id': retirement.id,
u'user': {
u'id': retirement.user.id,
u'username': retirement.user.username,
u'email': retirement.user.email,
u'profile': {
u'id': retirement.user.profile.id,
u'name': retirement.user.profile.name
},
},
u'original_username': retirement.original_username,
u'original_email': retirement.original_email,
u'original_name': retirement.original_name,
u'retired_username': retirement.retired_username,
u'retired_email': retirement.retired_email,
u'current_state': {
u'id': retirement.current_state.id,
u'state_name': retirement.current_state.state_name,
u'state_execution_order': retirement.current_state.state_execution_order,
},
u'last_state': {
u'id': retirement.last_state.id,
u'state_name': retirement.last_state.state_name,
u'state_execution_order': retirement.last_state.state_execution_order,
},
u'created': retirement.created,
u'modified': retirement.modified
}
if all_fields:
retirement_dict['responses'] = retirement.responses
return retirement_dict
def _create_users_all_states(self):
return [self._create_retirement(state) for state in RetirementState.objects.all()]
def _get_non_dead_end_states(self):
return [state for state in RetirementState.objects.filter(is_dead_end_state=False)]
def _get_dead_end_states(self):
return [state for state in RetirementState.objects.filter(is_dead_end_state=True)]
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestDeactivateLogout(RetirementTestCase):
"""
Tests the account deactivation/logout endpoint.
"""
def setUp(self):
super(TestDeactivateLogout, self).setUp()
self.test_password = 'password'
self.test_user = UserFactory(password=self.test_password)
UserSocialAuth.objects.create(
user=self.test_user,
provider='some_provider_name',
uid='xyz@gmail.com'
)
UserSocialAuth.objects.create(
user=self.test_user,
provider='some_other_provider_name',
uid='xyz@gmail.com'
)
Registration().register(self.test_user)
self.url = reverse('deactivate_logout')
def build_post(self, password):
return {'password': password}
@mock.patch('openedx.core.djangolib.oauth2_retirement_utils')
def test_user_can_deactivate_self(self, retirement_utils_mock):
"""
Verify a user calling the deactivation endpoint logs out the user, deletes all their SSO tokens,
and creates a user retirement row.
"""
self.client.login(username=self.test_user.username, password=self.test_password)
headers = build_jwt_headers(self.test_user)
response = self.client.post(self.url, self.build_post(self.test_password), **headers)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
# make sure the user model is as expected
updated_user = User.objects.get(id=self.test_user.id)
self.assertEqual(get_retired_email_by_email(self.test_user.email), updated_user.email)
self.assertFalse(updated_user.has_usable_password())
self.assertEqual(list(UserSocialAuth.objects.filter(user=self.test_user)), [])
self.assertEqual(list(Registration.objects.filter(user=self.test_user)), [])
self.assertEqual(len(UserRetirementStatus.objects.filter(user_id=self.test_user.id)), 1)
# these retirement utils are tested elsewhere; just make sure we called them
retirement_utils_mock.retire_dop_oauth2_models.assertCalledWith(self.test_user)
retirement_utils_mock.retire_dot_oauth2_models.assertCalledWith(self.test_user)
# make sure the user cannot log in
self.assertFalse(self.client.login(username=self.test_user.username, password=self.test_password))
# make sure that an email has been sent
self.assertEqual(len(mail.outbox), 1)
# ensure that it's been sent to the correct email address
self.assertIn(self.test_user.email, mail.outbox[0].to)
def test_password_mismatch(self):
"""
Verify that the user submitting a mismatched password results in
a rejection.
"""
self.client.login(username=self.test_user.username, password=self.test_password)
headers = build_jwt_headers(self.test_user)
response = self.client.post(self.url, self.build_post(self.test_password + "xxxx"), **headers)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_called_twice(self):
"""
Verify a user calling the deactivation endpoint a second time results in a "forbidden"
error, as the user will be logged out.
"""
self.client.login(username=self.test_user.username, password=self.test_password)
headers = build_jwt_headers(self.test_user)
response = self.client.post(self.url, self.build_post(self.test_password), **headers)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.client.login(username=self.test_user.username, password=self.test_password)
headers = build_jwt_headers(self.test_user)
response = self.client.post(self.url, self.build_post(self.test_password), **headers)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestAccountRetireMailings(RetirementTestCase):
"""
Tests the account retire mailings endpoint.
"""
def setUp(self):
super(TestAccountRetireMailings, self).setUp()
self.test_superuser = SuperuserFactory()
self.test_service_user = UserFactory()
# Should be created in parent setUpClass
retiring_email_lists = RetirementState.objects.get(state_name='RETIRING_EMAIL_LISTS')
self.retirement = self._create_retirement(retiring_email_lists)
self.test_user = self.retirement.user
UserOrgTag.objects.create(user=self.test_user, key='email-optin', org="foo", value="True")
UserOrgTag.objects.create(user=self.test_user, key='email-optin', org="bar", value="True")
self.url = reverse('accounts_retire_mailings')
def build_post(self, user):
return {'username': user.username}
def assert_status_and_tag_count(self, headers, expected_status=status.HTTP_204_NO_CONTENT, expected_tag_count=2,
expected_tag_value="False", expected_content=None):
"""
Helper function for making a request to the retire subscriptions endpoint, and asserting the status.
"""
response = self.client.post(self.url, self.build_post(self.test_user), **headers)
self.assertEqual(response.status_code, expected_status)
# Check that the expected number of tags with the correct value exist
tag_count = UserOrgTag.objects.filter(user=self.test_user, value=expected_tag_value).count()
self.assertEqual(tag_count, expected_tag_count)
if expected_content:
self.assertEqual(response.content.strip('"'), expected_content)
def test_superuser_retires_user_subscriptions(self):
"""
Verify a user's subscriptions are retired when a superuser posts to the retire subscriptions endpoint.
"""
headers = build_jwt_headers(self.test_superuser)
self.assert_status_and_tag_count(headers)
def test_superuser_retires_user_subscriptions_no_orgtags(self):
"""
Verify the call succeeds when the user doesn't have any org tags.
"""
UserOrgTag.objects.all().delete()
headers = build_jwt_headers(self.test_superuser)
self.assert_status_and_tag_count(headers, expected_tag_count=0)
def test_unauthorized_rejection(self):
"""
Verify unauthorized users cannot retire subscriptions.
"""
headers = build_jwt_headers(self.test_user)
# User should still have 2 "True" subscriptions.
self.assert_status_and_tag_count(headers, expected_status=status.HTTP_403_FORBIDDEN, expected_tag_value="True")
def test_signal_failure(self):
"""
Verify that if a signal fails the transaction is rolled back and a proper error message is returned.
"""
headers = build_jwt_headers(self.test_superuser)
mock_handler = mock.MagicMock()
mock_handler.side_effect = Exception("Tango")
try:
USER_RETIRE_MAILINGS.connect(mock_handler)
# User should still have 2 "True" subscriptions.
self.assert_status_and_tag_count(
headers,
expected_status=status.HTTP_500_INTERNAL_SERVER_ERROR,
expected_tag_value="True",
expected_content="Tango"
)
finally:
USER_RETIRE_MAILINGS.disconnect(mock_handler)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestPartnerReportingCleanup(ModuleStoreTestCase):
"""
Tests the partner reporting cleanup endpoint.
"""
def setUp(self):
super(TestPartnerReportingCleanup, self).setUp()
self.test_superuser = SuperuserFactory()
self.course = CourseFactory()
self.course_awesome_org = CourseFactory(org='awesome_org')
self.headers = build_jwt_headers(self.test_superuser)
self.headers['content_type'] = "application/json"
self.url = reverse('accounts_retirement_partner_report')
self.maxDiff = None
def create_partner_reporting_statuses(self, is_being_processed=True, num=2):
"""
Creates and returnes the given number of test users and UserRetirementPartnerReportingStatuses
with the given is_being_processed value.
"""
statuses = []
for _ in range(num):
user = UserFactory()
reporting_status = UserRetirementPartnerReportingStatus.objects.create(
user=user,
original_username=user.username,
original_email=user.email,
original_name=user.first_name + ' ' + user.last_name,
is_being_processed=is_being_processed
)
statuses.append(reporting_status)
return statuses
def assert_status_and_count(self, statuses, remaining_count, expected_status=status.HTTP_204_NO_CONTENT):
"""
Performs a test client DELETE against the retirement reporting cleanup endpoint. It generates
the JSON of usernames to clean up based on the given list of UserRetirementPartnerReportingStatuses,
asserts that the given number of UserRetirementPartnerReportingStatus rows are still in the database
after the operation, and asserts that the given expected_status HTTP status code is returned.
"""
usernames = [{'original_username': u.original_username} for u in statuses]
data = json.dumps(usernames)
response = self.client.delete(self.url, data=data, **self.headers)
print(response)
print(response.content)
self.assertEqual(response.status_code, expected_status)
self.assertEqual(UserRetirementPartnerReportingStatus.objects.all().count(), remaining_count)
def test_success(self):
"""
A basic test that newly created UserRetirementPartnerReportingStatus rows are all deleted as expected.
"""
statuses = self.create_partner_reporting_statuses()
self.assert_status_and_count(statuses, remaining_count=0)
def test_no_usernames(self):
"""
Checks that if no usernames are passed in we will get a 400 back.
"""
statuses = self.create_partner_reporting_statuses()
self.assert_status_and_count([], len(statuses), expected_status=status.HTTP_400_BAD_REQUEST)
def test_username_does_not_exist(self):
"""
Checks that if a username is passed in that does not have a UserRetirementPartnerReportingStatus row
we will get a 400 back.
"""
statuses = self.create_partner_reporting_statuses()
orig_count = len(statuses)
# Create a bogus user that has a non-saved row. This user doesn't exist in the database, so
# it should trigger the "incorrect number of rows" error.
user = UserFactory()
statuses.append(
UserRetirementPartnerReportingStatus(
user=user,
original_username=user.username,
original_email=user.email,
original_name=user.first_name + ' ' + user.last_name,
is_being_processed=True
)
)
self.assert_status_and_count(statuses, orig_count, expected_status=status.HTTP_400_BAD_REQUEST)
def test_username_in_wrong_status(self):
"""
Checks that if an username passed in has the wrong "is_being_processed" value we will get a 400 error.
"""
# Create some status rows in the expected status
statuses = self.create_partner_reporting_statuses()
# Create some users in the wrong processing status, should trigger the "incorrect number of rows" error.
statuses += self.create_partner_reporting_statuses(is_being_processed=False)
self.assert_status_and_count(statuses, len(statuses), expected_status=status.HTTP_400_BAD_REQUEST)
def test_does_not_delete_users_in_process(self):
"""
Checks that with mixed "is_being_processed" values in the table only the usernames passed in will
be deleted.
"""
statuses = self.create_partner_reporting_statuses()
self.create_partner_reporting_statuses(is_being_processed=False)
self.assert_status_and_count(statuses, len(statuses))
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestPartnerReportingList(ModuleStoreTestCase):
"""
Tests the partner reporting list endpoint
"""
def setUp(self):
super(TestPartnerReportingList, self).setUp()
self.test_superuser = SuperuserFactory()
self.course = CourseFactory()
self.course_awesome_org = CourseFactory(org='awesome_org')
self.courses = (self.course, self.course_awesome_org)
self.headers = build_jwt_headers(self.test_superuser)
self.url = reverse('accounts_retirement_partner_report')
self.maxDiff = None
@staticmethod
def get_user_dict(user, enrollments):
"""
Emulate the DRF serialization to create a dict we can compare against the partner
reporting list endpoint results. If this breaks in testing the serialization will
have changed and clients of this endpoint will need to be updates as well.
"""
return {
'original_username': user.username,
'original_email': user.email,
'original_name': user.first_name + ' ' + user.last_name,
'orgs': [enrollment.course.org for enrollment in enrollments]
}
def create_partner_reporting_statuses(self, is_being_processed=False, num=2, courses=None):
"""
Create the given number of test users and UserRetirementPartnerReportingStatus rows,
enroll them in the given course (or the default test course if none given), and set
their processing state to "is_being_processed".
Returns a list of user dicts representing what we would expect back from the
endpoint for the given user / enrollment.
"""
user_dicts = []
courses = self.courses if courses is None else courses
for _ in range(num):
user = UserFactory()
UserRetirementPartnerReportingStatus.objects.create(
user=user,
original_username=user.username,
original_email=user.email,
original_name=user.first_name + ' ' + user.last_name,
is_being_processed=is_being_processed
)
enrollments = []
for course in courses:
enrollments.append(CourseEnrollment.enroll(user=user, course_key=course.id))
user_dicts.append(
self.get_user_dict(user, enrollments)
)
return user_dicts
def assert_status_and_user_list(self, expected_users, expected_status=status.HTTP_200_OK):
"""
Makes the partner reporting list GET and asserts that the given users are
in the returned list, as well as asserting the expected HTTP status code
is returned.
"""
response = self.client.post(self.url, **self.headers)
self.assertEqual(response.status_code, expected_status)
returned_users = response.json()
print(returned_users)
print(expected_users)
self.assertEqual(len(expected_users), len(returned_users))
# These sub-lists will fail assertCountEqual if they're out of order
for expected_user in expected_users:
expected_user['orgs'].sort()
for returned_user in returned_users:
returned_user['orgs'].sort()
self.assertCountEqual(returned_users, expected_users)
def test_success(self):
"""
Basic test to make sure that users in two different orgs are returned.
"""
users = self.create_partner_reporting_statuses()
users += self.create_partner_reporting_statuses(courses=(self.course_awesome_org,))
self.assert_status_and_user_list(users)
def test_success_multiple_statuses(self):
"""
Checks that only users in the correct is_being_processed state (False) are returned.
"""
users = self.create_partner_reporting_statuses()
# These should not come back
self.create_partner_reporting_statuses(courses=(self.course_awesome_org,), is_being_processed=True)
self.assert_status_and_user_list(users)
def test_no_users(self):
"""
Checks that the call returns a success code and empty list if no users are found.
"""
self.assert_status_and_user_list([])
def test_only_users_in_processing(self):
"""
Checks that the call returns a success code and empty list if only users with
"is_being_processed=True" are in the database.
"""
self.create_partner_reporting_statuses(is_being_processed=True)
self.assert_status_and_user_list([])
def test_state_update(self):
"""
Checks that users are progressed to "is_being_processed" True upon being returned
from this call.
"""
users = self.create_partner_reporting_statuses()
# First time through we should get the users
self.assert_status_and_user_list(users)
# Second time they should be updated to is_being_processed=True
self.assert_status_and_user_list([])
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestAccountRetirementList(RetirementTestCase):
"""
Tests the account retirement endpoint.
"""
def setUp(self):
super(TestAccountRetirementList, self).setUp()
self.test_superuser = SuperuserFactory()
self.headers = build_jwt_headers(self.test_superuser)
self.url = reverse('accounts_retirement_queue')
self.maxDiff = None
def assert_status_and_user_list(
self,
expected_data,
expected_status=status.HTTP_200_OK,
states_to_request=None,
cool_off_days=7
):
"""
Helper function for making a request to the retire subscriptions endpoint, asserting the status, and
optionally asserting data returned.
"""
if states_to_request is None:
# These are just a couple of random states that should be used in any implementation
states_to_request = ['PENDING', 'LOCKING_ACCOUNT']
else:
# Can pass in RetirementState objects or strings here
try:
states_to_request = [s.state_name for s in states_to_request]
except AttributeError:
states_to_request = states_to_request
data = {'cool_off_days': cool_off_days, 'states': states_to_request}
response = self.client.get(self.url, data, **self.headers)
self.assertEqual(response.status_code, expected_status)
response_data = response.json()
if expected_data:
# These datetimes won't match up due to serialization, but they're inherited fields tested elsewhere
for data in (response_data, expected_data):
for retirement in data:
del retirement['created']
del retirement['modified']
self.assertItemsEqual(response_data, expected_data)
def test_empty(self):
"""
Verify that an empty array is returned if no users are awaiting retirement
"""
self.assert_status_and_user_list([])
def test_users_exist_none_in_correct_status(self):
"""
Verify that users in dead end states are not returned
"""
for state in self._get_dead_end_states():
self._create_retirement(state)
self.assert_status_and_user_list([], states_to_request=self._get_non_dead_end_states())
def test_users_retrieved_in_multiple_states(self):
"""
Verify that if multiple states are requested, learners in each state are returned.
"""
multiple_states = ['PENDING', 'FORUMS_COMPLETE']
for state in multiple_states:
self._create_retirement(RetirementState.objects.get(state_name=state))
data = {'cool_off_days': 0, 'states': multiple_states}
response = self.client.get(self.url, data, **self.headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.json()), 2)
def test_users_exist(self):
"""
Verify users in different states are returned with correct data or filtered out
"""
self.maxDiff = None
retirement_values = []
states_to_request = []
dead_end_states = self._get_dead_end_states()
for retirement in self._create_users_all_states():
if retirement.current_state not in dead_end_states:
states_to_request.append(retirement.current_state)
retirement_values.append(self._retirement_to_dict(retirement))
self.assert_status_and_user_list(retirement_values, states_to_request=self._get_non_dead_end_states())
def test_date_filter(self):
"""
Verifies the functionality of the `cool_off_days` parameter by creating 1 retirement per day for
10 days. Then requests different 1-10 `cool_off_days` to confirm the correct retirements are returned.
"""
retirements = []
days_back_to_test = 10
# Create a retirement per day for the last 10 days, from oldest date to newest. We want these all created
# before we start checking, thus the two loops.
# retirements = [2018-04-10..., 2018-04-09..., 2018-04-08...]
pending_state = RetirementState.objects.get(state_name='PENDING')
for days_back in range(1, days_back_to_test, -1):
create_datetime = datetime.datetime.now(pytz.UTC) - datetime.timedelta(days=days_back)
retirements.append(self._create_retirement(state=pending_state, create_datetime=create_datetime))
# Confirm we get the correct number and data back for each day we add to cool off days
# For each day we add to `cool_off_days` we expect to get one fewer retirement.
for cool_off_days in range(1, days_back_to_test):
# Start with 9 days back
req_days_back = days_back_to_test - cool_off_days
retirement_dicts = [self._retirement_to_dict(ret) for ret in retirements[:cool_off_days]]
self.assert_status_and_user_list(
retirement_dicts,
cool_off_days=req_days_back
)
def test_bad_cool_off_days(self):
"""
Check some bad inputs to make sure we get back the expected status
"""
self.assert_status_and_user_list(None, expected_status=status.HTTP_400_BAD_REQUEST, cool_off_days=-1)
self.assert_status_and_user_list(None, expected_status=status.HTTP_400_BAD_REQUEST, cool_off_days='ABCDERTP')
def test_bad_states(self):
"""
Check some bad inputs to make sure we get back the expected status
"""
self.assert_status_and_user_list(
None,
expected_status=status.HTTP_400_BAD_REQUEST,
states_to_request=['TUNA', 'TACO'])
self.assert_status_and_user_list(None, expected_status=status.HTTP_400_BAD_REQUEST, states_to_request=[])
def test_missing_params(self):
"""
All params are required, make sure that is enforced
"""
response = self.client.get(self.url, **self.headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
response = self.client.get(self.url, {}, **self.headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
response = self.client.get(self.url, {'cool_off_days': 7}, **self.headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
RetirementState.objects.get(state_name='PENDING')
response = self.client.get(self.url, {'states': ['PENDING']}, **self.headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestAccountRetirementRetrieve(RetirementTestCase):
"""
Tests the account retirement retrieval endpoint.
"""
def setUp(self):
super(TestAccountRetirementRetrieve, self).setUp()
self.test_user = UserFactory()
self.test_superuser = SuperuserFactory()
self.url = reverse('accounts_retirement_retrieve', kwargs={'username': self.test_user.username})
self.headers = build_jwt_headers(self.test_superuser)
self.maxDiff = None
def assert_status_and_user_data(self, expected_data, expected_status=status.HTTP_200_OK, username_to_find=None):
"""
Helper function for making a request to the retire subscriptions endpoint, asserting the status,
and optionally asserting the expected data.
"""
if username_to_find is not None:
self.url = reverse('accounts_retirement_retrieve', kwargs={'username': username_to_find})
response = self.client.get(self.url, **self.headers)
self.assertEqual(response.status_code, expected_status)
if expected_data is not None:
response_data = response.json()
# These won't match up due to serialization, but they're inherited fields tested elsewhere
for data in (expected_data, response_data):
del data['created']
del data['modified']
self.assertDictEqual(response_data, expected_data)
return response_data
def test_no_retirement(self):
"""
Confirm we get a 404 if a retirement for the user can be found
"""
self.assert_status_and_user_data(None, status.HTTP_404_NOT_FOUND)
def test_retirements_all_states(self):
"""
Create a bunch of retirements and confirm we get back the correct data for each
"""
retirements = []
for state in RetirementState.objects.all():
retirements.append(self._create_retirement(state))
for retirement in retirements:
values = self._retirement_to_dict(retirement)
self.assert_status_and_user_data(values, username_to_find=values['user']['username'])
def test_retrieve_by_old_username(self):
"""
Simulate retrieving a retirement by the old username, after the name has been changed to the hashed one
"""
pending_state = RetirementState.objects.get(state_name='PENDING')
retirement = self._create_retirement(pending_state)
original_username = retirement.user.username
hashed_username = get_retired_username_by_username(original_username)
retirement.user.username = hashed_username
retirement.user.save()
values = self._retirement_to_dict(retirement)
self.assert_status_and_user_data(values, username_to_find=original_username)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestAccountRetirementUpdate(RetirementTestCase):
"""
Tests the account retirement endpoint.
"""
def setUp(self):
super(TestAccountRetirementUpdate, self).setUp()
self.pending_state = RetirementState.objects.get(state_name='PENDING')
self.locking_state = RetirementState.objects.get(state_name='LOCKING_ACCOUNT')
self.retirement = self._create_retirement(self.pending_state)
self.test_user = self.retirement.user
self.test_superuser = SuperuserFactory()
self.headers = build_jwt_headers(self.test_superuser)
self.headers['content_type'] = "application/json"
self.url = reverse('accounts_retirement_update')
def update_and_assert_status(self, data, expected_status=status.HTTP_204_NO_CONTENT):
"""
Helper function for making a request to the retire subscriptions endpoint, and asserting the status.
"""
if 'username' not in data:
data['username'] = self.test_user.username
response = self.client.patch(self.url, json.dumps(data), **self.headers)
self.assertEqual(response.status_code, expected_status)
def test_single_update(self):
"""
Basic test to confirm changing state works and saves the given response
"""
data = {'new_state': 'LOCKING_ACCOUNT', 'response': 'this should succeed'}
self.update_and_assert_status(data)
# Refresh the retirment object and confirm the messages and state are correct
retirement = UserRetirementStatus.objects.get(id=self.retirement.id)
self.assertEqual(retirement.current_state, RetirementState.objects.get(state_name='LOCKING_ACCOUNT'))
self.assertEqual(retirement.last_state, RetirementState.objects.get(state_name='PENDING'))
self.assertIn('this should succeed', retirement.responses)
def test_move_through_process(self):
"""
Simulate moving a retirement through the process and confirm they end up in the
correct state, with all relevant response messages logged.
"""
fake_retire_process = [
{'new_state': 'LOCKING_ACCOUNT', 'response': 'accountlockstart'},
{'new_state': 'LOCKING_COMPLETE', 'response': 'accountlockcomplete'},
{'new_state': 'RETIRING_CREDENTIALS', 'response': 'retiringcredentials'},
{'new_state': 'CREDENTIALS_COMPLETE', 'response': 'credentialsretired'},
{'new_state': 'COMPLETE', 'response': 'accountretirementcomplete'},
]
for update_data in fake_retire_process:
self.update_and_assert_status(update_data)
# Refresh the retirment object and confirm the messages and state are correct
retirement = UserRetirementStatus.objects.get(id=self.retirement.id)
self.assertEqual(retirement.current_state, RetirementState.objects.get(state_name='COMPLETE'))
self.assertEqual(retirement.last_state, RetirementState.objects.get(state_name='CREDENTIALS_COMPLETE'))
self.assertIn('accountlockstart', retirement.responses)
self.assertIn('accountlockcomplete', retirement.responses)
self.assertIn('retiringcredentials', retirement.responses)
self.assertIn('credentialsretired', retirement.responses)
self.assertIn('accountretirementcomplete', retirement.responses)
def test_unknown_state(self):
"""
Test that trying to set to an unknown state fails with a 400
"""
data = {'new_state': 'BOGUS_STATE', 'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
def test_bad_vars(self):
"""
Test various ways of sending the wrong variables to make sure they all fail correctly
"""
# No `new_state`
data = {'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
# No `response`
data = {'new_state': 'COMPLETE'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
# Unknown `new_state`
data = {'new_state': 'BOGUS_STATE', 'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
# No `new_state` or `response`
data = {}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
# Unexpected param `should_not_exist`
data = {'should_not_exist': 'bad', 'new_state': 'COMPLETE', 'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
def test_no_retirement(self):
"""
Confirm that trying to operate on a non-existent retirement for an existing user 404s
"""
# Delete the only retirement, created in setUp
UserRetirementStatus.objects.all().delete()
data = {'new_state': 'LOCKING_ACCOUNT', 'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_404_NOT_FOUND)
def test_no_user(self):
"""
Confirm that trying to operate on a non-existent user 404s
"""
data = {'new_state': 'LOCKING_ACCOUNT', 'response': 'this should fail', 'username': 'does not exist'}
self.update_and_assert_status(data, status.HTTP_404_NOT_FOUND)
def test_move_from_dead_end(self):
"""
Confirm that trying to move from a dead end state to any other state fails
"""
retirement = UserRetirementStatus.objects.get(id=self.retirement.id)
retirement.current_state = RetirementState.objects.filter(is_dead_end_state=True)[0]
retirement.save()
data = {'new_state': 'LOCKING_ACCOUNT', 'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
def test_move_backward(self):
"""
Confirm that trying to move to an earlier step in the process fails
"""
retirement = UserRetirementStatus.objects.get(id=self.retirement.id)
retirement.current_state = RetirementState.objects.get(state_name='COMPLETE')
retirement.save()
data = {'new_state': 'PENDING', 'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
def test_move_same(self):
"""
Confirm that trying to move to the same step in the process fails
"""
# Should already be in 'PENDING'
data = {'new_state': 'PENDING', 'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestAccountRetirementPost(RetirementTestCase):
"""
Tests the account retirement endpoint.
"""
def setUp(self):
super(TestAccountRetirementPost, self).setUp()
self.test_user = UserFactory()
self.test_superuser = SuperuserFactory()
self.original_username = self.test_user.username
self.original_email = self.test_user.email
self.retired_username = get_retired_username_by_username(self.original_username)
self.retired_email = get_retired_email_by_email(self.original_email)
retirement_state = RetirementState.objects.get(state_name='RETIRING_LMS')
self.retirement_status = UserRetirementStatus.create_retirement(self.test_user)
self.retirement_status.current_state = retirement_state
self.retirement_status.last_state = retirement_state
self.retirement_status.save()
SocialLink.objects.create(
user_profile=self.test_user.profile,
platform='Facebook',
social_link='www.facebook.com'
).save()
self.cache_key = UserProfile.country_cache_key_name(self.test_user.id)
cache.set(self.cache_key, 'Timor-leste')
# Enterprise model setup
self.course_id = 'course-v1:edX+DemoX.1+2T2017'
self.enterprise_customer = EnterpriseCustomer.objects.create(
name='test_enterprise_customer',
site=SiteFactory.create()
)
self.enterprise_user = EnterpriseCustomerUser.objects.create(
enterprise_customer=self.enterprise_customer,
user_id=self.test_user.id,
)
self.enterprise_enrollment = EnterpriseCourseEnrollment.objects.create(
enterprise_customer_user=self.enterprise_user,
course_id=self.course_id
)
self.pending_enterprise_user = PendingEnterpriseCustomerUser.objects.create(
enterprise_customer_id=self.enterprise_user.enterprise_customer_id,
user_email=self.test_user.email
)
self.sapsf_audit = SapSuccessFactorsLearnerDataTransmissionAudit.objects.create(
sapsf_user_id=self.test_user.id,
enterprise_course_enrollment_id=self.enterprise_enrollment.id,
completed_timestamp=1,
)
self.consent = DataSharingConsent.objects.create(
username=self.test_user.username,
enterprise_customer=self.enterprise_customer,
)
# Entitlement model setup
self.entitlement = CourseEntitlementFactory.create(user=self.test_user)
self.entitlement_support_detail = CourseEntitlementSupportDetail.objects.create(
entitlement=self.entitlement,
support_user=UserFactory(),
comments='A comment containing potential PII.'
)
# Misc. setup
self.photo_verification = SoftwareSecurePhotoVerificationFactory.create(user=self.test_user)
PendingEmailChangeFactory.create(user=self.test_user)
UserOrgTagFactory.create(user=self.test_user, key='foo', value='bar')
UserOrgTagFactory.create(user=self.test_user, key='cat', value='dog')
CourseEnrollmentAllowedFactory.create(email=self.original_email)
self.course_key = CourseKey.from_string('course-v1:edX+DemoX+Demo_Course')
self.cohort = CourseUserGroup.objects.create(
name="TestCohort",
course_id=self.course_key,
group_type=CourseUserGroup.COHORT
)
self.cohort_assignment = UnregisteredLearnerCohortAssignments.objects.create(
course_user_group=self.cohort,
course_id=self.course_key,
email=self.original_email
)
# setup for doing POST from test client
self.headers = build_jwt_headers(self.test_superuser)
self.headers['content_type'] = "application/json"
self.url = reverse('accounts_retire')
def post_and_assert_status(self, data, expected_status=status.HTTP_204_NO_CONTENT):
"""
Helper function for making a request to the retire subscriptions endpoint, and asserting the status.
"""
response = self.client.post(self.url, json.dumps(data), **self.headers)
self.assertEqual(response.status_code, expected_status)
return response
def test_user_profile_pii_has_expected_values(self):
expected_user_profile_pii = {
'name': '',
'meta': '',
'location': '',
'year_of_birth': None,
'gender': None,
'mailing_address': None,
'city': None,
'country': None,
'bio': None,
}
self.assertEqual(expected_user_profile_pii, USER_PROFILE_PII)
def test_retire_user_where_user_does_not_exist(self):
path = 'openedx.core.djangoapps.user_api.accounts.views.is_username_retired'
with mock.patch(path, return_value=False) as mock_retired_username:
data = {'username': 'not_a_user'}
response = self.post_and_assert_status(data, status.HTTP_404_NOT_FOUND)
self.assertFalse(response.content)
mock_retired_username.assert_called_once_with('not_a_user')
def test_retire_user_server_error_is_raised(self):
path = 'openedx.core.djangoapps.user_api.models.UserRetirementStatus.get_retirement_for_retirement_action'
with mock.patch(path, side_effect=Exception('Unexpected Exception')) as mock_get_retirement:
data = {'username': self.test_user.username}
response = self.post_and_assert_status(data, status.HTTP_500_INTERNAL_SERVER_ERROR)
self.assertEqual('Unexpected Exception', text_type(response.json()))
mock_get_retirement.assert_called_once_with(self.original_username)
def test_retire_user_where_user_already_retired(self):
path = 'openedx.core.djangoapps.user_api.accounts.views.is_username_retired'
with mock.patch(path, return_value=True) as mock_is_username_retired:
data = {'username': self.test_user.username}
response = self.post_and_assert_status(data, status.HTTP_404_NOT_FOUND)
self.assertFalse(response.content)
mock_is_username_retired.assert_called_once_with(self.original_username)
def test_retire_user_where_username_not_provided(self):
response = self.post_and_assert_status({}, status.HTTP_404_NOT_FOUND)
expected_response_message = {'message': text_type('The user was not specified.')}
self.assertEqual(expected_response_message, response.json())
@mock.patch('openedx.core.djangoapps.user_api.accounts.views.get_profile_image_names')
@mock.patch('openedx.core.djangoapps.user_api.accounts.views.remove_profile_images')
def test_retire_user(self, mock_remove_profile_images, mock_get_profile_image_names):
data = {'username': self.original_username}
self.post_and_assert_status(data)
self.test_user.refresh_from_db()
self.test_user.profile.refresh_from_db() # pylint: disable=no-member
expected_user_values = {
'first_name': '',
'last_name': '',
'is_active': False,
'username': self.retired_username,
}
for field, expected_value in expected_user_values.iteritems():
self.assertEqual(expected_value, getattr(self.test_user, field))
for field, expected_value in USER_PROFILE_PII.iteritems():
self.assertEqual(expected_value, getattr(self.test_user.profile, field))
self.assertIsNone(self.test_user.profile.profile_image_uploaded_at)
mock_get_profile_image_names.assert_called_once_with(self.original_username)
mock_remove_profile_images.assert_called_once_with(
mock_get_profile_image_names.return_value
)
self.assertFalse(
SocialLink.objects.filter(user_profile=self.test_user.profile).exists()
)
self.assertIsNone(cache.get(self.cache_key))
self._data_sharing_consent_assertions()
self._sapsf_audit_assertions()
self._pending_enterprise_customer_user_assertions()
self._entitlement_support_detail_assertions()
self._photo_verification_assertions()
self.assertFalse(PendingEmailChange.objects.filter(user=self.test_user).exists())
self.assertFalse(UserOrgTag.objects.filter(user=self.test_user).exists())
self.assertFalse(CourseEnrollmentAllowed.objects.filter(email=self.original_email).exists())
self.assertFalse(UnregisteredLearnerCohortAssignments.objects.filter(email=self.original_email).exists())
def test_deletes_pii_from_user_profile(self):
for model_field, value_to_assign in USER_PROFILE_PII.iteritems():
if value_to_assign == '':
value = 'foo'
else:
value = mock.Mock()
setattr(self.test_user.profile, model_field, value)
AccountRetirementView.clear_pii_from_userprofile(self.test_user)
for model_field, value_to_assign in USER_PROFILE_PII.iteritems():
self.assertEqual(value_to_assign, getattr(self.test_user.profile, model_field))
social_links = SocialLink.objects.filter(
user_profile=self.test_user.profile
)
self.assertFalse(social_links.exists())
@mock.patch('openedx.core.djangoapps.user_api.accounts.views.get_profile_image_names')
@mock.patch('openedx.core.djangoapps.user_api.accounts.views.remove_profile_images')
def test_removes_user_profile_images(
self, mock_remove_profile_images, mock_get_profile_image_names
):
test_datetime = datetime.datetime(2018, 1, 1)
self.test_user.profile.profile_image_uploaded_at = test_datetime
AccountRetirementView.delete_users_profile_images(self.test_user)
self.test_user.profile.refresh_from_db() # pylint: disable=no-member
self.assertIsNone(self.test_user.profile.profile_image_uploaded_at)
mock_get_profile_image_names.assert_called_once_with(self.test_user.username)
mock_remove_profile_images.assert_called_once_with(
mock_get_profile_image_names.return_value
)
def test_can_delete_user_profiles_country_cache(self):
AccountRetirementView.delete_users_country_cache(self.test_user)
self.assertIsNone(cache.get(self.cache_key))
def test_can_retire_users_datasharingconsent(self):
AccountRetirementView.retire_users_data_sharing_consent(self.test_user.username, self.retired_username)
self._data_sharing_consent_assertions()
def _data_sharing_consent_assertions(self):
"""
Helper method for asserting that ``DataSharingConsent`` objects are retired.
"""
self.consent.refresh_from_db()
self.assertEqual(self.retired_username, self.consent.username)
test_users_data_sharing_consent = DataSharingConsent.objects.filter(
username=self.original_username
)
self.assertFalse(test_users_data_sharing_consent.exists())
def test_can_retire_users_sap_success_factors_audits(self):
AccountRetirementView.retire_sapsf_data_transmission(self.test_user)
self._sapsf_audit_assertions()
def _sapsf_audit_assertions(self):
"""
Helper method for asserting that ``SapSuccessFactorsLearnerDataTransmissionAudit`` objects are retired.
"""
self.sapsf_audit.refresh_from_db()
self.assertEqual('', self.sapsf_audit.sapsf_user_id)
audits_for_original_user_id = SapSuccessFactorsLearnerDataTransmissionAudit.objects.filter(
sapsf_user_id=self.test_user.id,
)
self.assertFalse(audits_for_original_user_id.exists())
def test_can_retire_user_from_pendingenterprisecustomeruser(self):
AccountRetirementView.retire_user_from_pending_enterprise_customer_user(self.test_user, self.retired_email)
self._pending_enterprise_customer_user_assertions()
def _pending_enterprise_customer_user_assertions(self):
"""
Helper method for asserting that ``PendingEnterpriseCustomerUser`` objects are retired.
"""
self.pending_enterprise_user.refresh_from_db()
self.assertEqual(self.retired_email, self.pending_enterprise_user.user_email)
pending_enterprise_users = PendingEnterpriseCustomerUser.objects.filter(
user_email=self.original_email
)
self.assertFalse(pending_enterprise_users.exists())
def test_course_entitlement_support_detail_comments_are_retired(self):
AccountRetirementView.retire_entitlement_support_detail(self.test_user)
self._entitlement_support_detail_assertions()
def _entitlement_support_detail_assertions(self):
"""
Helper method for asserting that ``CourseEntitleSupportDetail`` objects are retired.
"""
self.entitlement_support_detail.refresh_from_db()
self.assertEqual('', self.entitlement_support_detail.comments)
def _photo_verification_assertions(self):
"""
Helper method for asserting that ``SoftwareSecurePhotoVerification`` objects are retired.
"""
self.photo_verification.refresh_from_db()
self.assertEqual(self.test_user, self.photo_verification.user)
for field in ('name', 'face_image_url', 'photo_id_image_url', 'photo_id_key'):
self.assertEqual('', getattr(self.photo_verification, field))
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestLMSAccountRetirementPost(RetirementTestCase, ModuleStoreTestCase):
"""
Tests the LMS account retirement (GDPR P2) endpoint.
"""
def setUp(self):
super(TestLMSAccountRetirementPost, self).setUp()
self.pii_standin = 'PII here'
self.course = CourseFactory()
self.test_user = UserFactory()
self.test_superuser = SuperuserFactory()
self.original_username = self.test_user.username
self.original_email = self.test_user.email
self.retired_username = get_retired_username_by_username(self.original_username)
self.retired_email = get_retired_email_by_email(self.original_email)
retirement_state = RetirementState.objects.get(state_name='RETIRING_LMS')
self.retirement_status = UserRetirementStatus.create_retirement(self.test_user)
self.retirement_status.current_state = retirement_state
self.retirement_status.last_state = retirement_state
self.retirement_status.save()
# wiki data setup
rp = RevisionPlugin.objects.create(article_id=0)
RevisionPluginRevision.objects.create(
revision_number=1,
ip_address="ipaddresss",
plugin=rp,
user=self.test_user,
)
article = Article.objects.create()
ArticleRevision.objects.create(ip_address="ipaddresss", user=self.test_user, article=article)
# ManualEnrollmentAudit setup
course_enrollment = CourseEnrollment.enroll(user=self.test_user, course_key=self.course.id)
ManualEnrollmentAudit.objects.create(
enrollment=course_enrollment, reason=self.pii_standin, enrolled_email=self.pii_standin
)
# CreditRequest and CreditRequirementStatus setup
provider = CreditProvider.objects.create(provider_id="Hogwarts")
credit_course = CreditCourse.objects.create(course_key=self.course.id)
CreditRequest.objects.create(
username=self.test_user.username,
course=credit_course,
provider_id=provider.id,
parameters={self.pii_standin},
)
req = CreditRequirement.objects.create(course_id=credit_course.id)
CreditRequirementStatus.objects.create(username=self.test_user.username, requirement=req)
# ApiAccessRequest setup
site = Site.objects.create()
ApiAccessRequest.objects.create(
user=self.test_user,
site=site,
website=self.pii_standin,
company_address=self.pii_standin,
company_name=self.pii_standin,
reason=self.pii_standin,
)
# SurveyAnswer setup
SurveyAnswer.objects.create(user=self.test_user, field_value=self.pii_standin, form_id=0)
# other setup
PendingNameChange.objects.create(user=self.test_user, new_name=self.pii_standin, rationale=self.pii_standin)
PasswordHistory.objects.create(user=self.test_user, password=self.pii_standin)
# setup for doing POST from test client
self.headers = build_jwt_headers(self.test_superuser)
self.headers['content_type'] = "application/json"
self.url = reverse('accounts_retire_misc')
def post_and_assert_status(self, data, expected_status=status.HTTP_204_NO_CONTENT):
"""
Helper function for making a request to the retire subscriptions endpoint, and asserting the status.
"""
response = self.client.post(self.url, json.dumps(data), **self.headers)
self.assertEqual(response.status_code, expected_status)
return response
def test_retire_user(self):
# check that rows that will not exist after retirement exist now
self.assertTrue(CreditRequest.objects.filter(username=self.test_user.username).exists())
self.assertTrue(CreditRequirementStatus.objects.filter(username=self.test_user.username).exists())
self.assertTrue(PendingNameChange.objects.filter(user=self.test_user).exists())
retirement = UserRetirementStatus.get_retirement_for_retirement_action(self.test_user.username)
data = {'username': self.original_username}
self.post_and_assert_status(data)
self.test_user.refresh_from_db()
self.test_user.profile.refresh_from_db() # pylint: disable=no-member
self.assertEqual(RevisionPluginRevision.objects.get(user=self.test_user).ip_address, None)
self.assertEqual(ArticleRevision.objects.get(user=self.test_user).ip_address, None)
self.assertFalse(PendingNameChange.objects.filter(user=self.test_user).exists())
self.assertEqual(PasswordHistory.objects.get(user=self.test_user).password, '')
self.assertEqual(
ManualEnrollmentAudit.objects.get(
enrollment=CourseEnrollment.objects.get(user=self.test_user)
).enrolled_email,
retirement.retired_email
)
self.assertFalse(CreditRequest.objects.filter(username=self.test_user.username).exists())
self.assertTrue(CreditRequest.objects.filter(username=retirement.retired_username).exists())
self.assertEqual(CreditRequest.objects.get(username=retirement.retired_username).parameters, {})
self.assertFalse(CreditRequirementStatus.objects.filter(username=self.test_user.username).exists())
self.assertTrue(CreditRequirementStatus.objects.filter(username=retirement.retired_username).exists())
self.assertEqual(CreditRequirementStatus.objects.get(username=retirement.retired_username).reason, {})
retired_api_access_request = ApiAccessRequest.objects.get(user=self.test_user)
self.assertEqual(retired_api_access_request.website, '')
self.assertEqual(retired_api_access_request.company_address, '')
self.assertEqual(retired_api_access_request.company_name, '')
self.assertEqual(retired_api_access_request.reason, '')
self.assertEqual(SurveyAnswer.objects.get(user=self.test_user).field_value, '')

1215
openedx/core/djangoapps/user_api/accounts/tests/test_views.py
View File

@@ -6,57 +6,22 @@ from copy import deepcopy
import datetime
import hashlib
import json
import unittest
from consent.models import DataSharingConsent
import ddt
from django.conf import settings
from django.contrib.auth.models import User
from django.contrib.sites.models import Site
from django.core import mail
from django.core.cache import cache
from django.urls import reverse
from django.test import TestCase
from django.core.urlresolvers import reverse
from django.test.testcases import TransactionTestCase
from django.test.utils import override_settings
from enterprise.models import (
EnterpriseCustomer,
EnterpriseCustomerUser,
EnterpriseCourseEnrollment,
PendingEnterpriseCustomerUser,
)
from integrated_channels.sap_success_factors.models import (
SapSuccessFactorsLearnerDataTransmissionAudit
)
import mock
from nose.plugins.attrib import attr
from opaque_keys.edx.keys import CourseKey
import pytz
from rest_framework import status
from rest_framework.test import APIClient, APITestCase
from six import text_type
from social_django.models import UserSocialAuth
from wiki.models import ArticleRevision, Article
from wiki.models.pluginbase import RevisionPluginRevision, RevisionPlugin
from xmodule.modulestore.tests.factories import CourseFactory
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from entitlements.models import CourseEntitlementSupportDetail
from entitlements.tests.factories import CourseEntitlementFactory
from lms.djangoapps.verify_student.tests.factories import SoftwareSecurePhotoVerificationFactory
from openedx.core.djangoapps.api_admin.models import ApiAccessRequest
from openedx.core.djangoapps.credit.models import (
CreditRequirementStatus, CreditRequest, CreditCourse, CreditProvider, CreditRequirement
)
from openedx.core.djangoapps.course_groups.models import CourseUserGroup, UnregisteredLearnerCohortAssignments
from openedx.core.djangoapps.site_configuration.tests.factories import SiteFactory
from openedx.core.djangoapps.user_api.accounts import ACCOUNT_VISIBILITY_PREF_KEY
from openedx.core.djangoapps.user_api.accounts.signals import USER_RETIRE_MAILINGS
from openedx.core.djangoapps.user_api.models import RetirementState, UserRetirementStatus, UserPreference, UserOrgTag
from openedx.core.djangoapps.user_api.models import UserPreference
from openedx.core.djangoapps.user_api.preferences.api import set_user_preference
from openedx.core.djangolib.testing.utils import CacheIsolationTestCase, skip_unless_lms
from openedx.core.lib.token_utils import JwtBuilder
from survey.models import SurveyAnswer
from student.models import (
CourseEnrollment,
CourseEnrollmentAllowed,
@@ -81,8 +46,7 @@ from student.tests.factories import (
)
from .. import ALL_USERS_VISIBILITY, PRIVATE_VISIBILITY
from ..views import AccountRetirementView, USER_PROFILE_PII
from ...tests.factories import UserOrgTagFactory
TEST_PROFILE_IMAGE_UPLOADED_AT = datetime.datetime(2002, 1, 9, 15, 43, 1, tzinfo=pytz.UTC)
@@ -882,1174 +846,3 @@ class TestAccountAPITransactions(TransactionTestCase):
data = response.data
self.assertEqual(old_email, data["email"])
self.assertEqual(u"m", data["gender"])
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestAccountDeactivation(TestCase):
"""
Tests the account deactivation endpoint.
"""
def setUp(self):
super(TestAccountDeactivation, self).setUp()
self.test_user = UserFactory()
self.url = reverse('accounts_deactivation', kwargs={'username': self.test_user.username})
def build_jwt_headers(self, user):
"""
Helper function for creating headers for the JWT authentication.
"""
token = JwtBuilder(user).build_token([])
headers = {
'HTTP_AUTHORIZATION': 'JWT ' + token
}
return headers
def assert_activation_status(self, headers, expected_status=status.HTTP_200_OK, expected_activation_status=False):
"""
Helper function for making a request to the deactivation endpoint, and asserting the status.
Args:
expected_status(int): Expected request's response status.
expected_activation_status(bool): Expected user has_usable_password attribute value.
"""
self.assertTrue(self.test_user.has_usable_password())
response = self.client.post(self.url, **headers)
self.assertEqual(response.status_code, expected_status)
self.test_user.refresh_from_db()
self.assertEqual(self.test_user.has_usable_password(), expected_activation_status)
def test_superuser_deactivates_user(self):
"""
Verify a user is deactivated when a superuser posts to the deactivation endpoint.
"""
superuser = SuperuserFactory()
headers = self.build_jwt_headers(superuser)
self.assert_activation_status(headers)
def test_user_with_permission_deactivates_user(self):
"""
Verify a user is deactivated when a user with permission posts to the deactivation endpoint.
"""
user = UserFactory()
permission = PermissionFactory(
codename='can_deactivate_users',
content_type=ContentTypeFactory(
app_label='student'
)
)
user.user_permissions.add(permission)
headers = self.build_jwt_headers(user)
self.assertTrue(self.test_user.has_usable_password())
self.assert_activation_status(headers)
def test_unauthorized_rejection(self):
"""
Verify unauthorized users cannot deactivate accounts.
"""
headers = self.build_jwt_headers(self.test_user)
self.assert_activation_status(
headers,
expected_status=status.HTTP_403_FORBIDDEN,
expected_activation_status=True
)
def test_on_jwt_headers_rejection(self):
"""
Verify users who are not JWT authenticated are rejected.
"""
UserFactory()
self.assert_activation_status(
{},
expected_status=status.HTTP_401_UNAUTHORIZED,
expected_activation_status=True
)
class RetirementTestCase(TestCase):
"""
Test case with a helper methods for retirement
"""
@classmethod
def setUpClass(cls):
super(RetirementTestCase, cls).setUpClass()
cls.setup_states()
@staticmethod
def setup_states():
"""
Create basic states that mimic our current understanding of the retirement process
"""
default_states = [
('PENDING', 1, False, True),
('LOCKING_ACCOUNT', 20, False, False),
('LOCKING_COMPLETE', 30, False, False),
('RETIRING_CREDENTIALS', 40, False, False),
('CREDENTIALS_COMPLETE', 50, False, False),
('RETIRING_ECOM', 60, False, False),
('ECOM_COMPLETE', 70, False, False),
('RETIRING_FORUMS', 80, False, False),
('FORUMS_COMPLETE', 90, False, False),
('RETIRING_EMAIL_LISTS', 100, False, False),
('EMAIL_LISTS_COMPLETE', 110, False, False),
('RETIRING_ENROLLMENTS', 120, False, False),
('ENROLLMENTS_COMPLETE', 130, False, False),
('RETIRING_NOTES', 140, False, False),
('NOTES_COMPLETE', 150, False, False),
('NOTIFYING_PARTNERS', 160, False, False),
('PARTNERS_NOTIFIED', 170, False, False),
('RETIRING_LMS', 180, False, False),
('LMS_COMPLETE', 190, False, False),
('ERRORED', 200, True, True),
('ABORTED', 210, True, True),
('COMPLETE', 220, True, True),
]
for name, ex, dead, req in default_states:
RetirementState.objects.create(
state_name=name,
state_execution_order=ex,
is_dead_end_state=dead,
required=req
)
def build_jwt_headers(self, user):
"""
Helper function for creating headers for the JWT authentication.
"""
token = JwtBuilder(user).build_token([])
headers = {
'HTTP_AUTHORIZATION': 'JWT ' + token
}
return headers
def _create_retirement(self, state, create_datetime=None):
"""
Helper method to create a RetirementStatus with useful defaults
"""
if create_datetime is None:
create_datetime = datetime.datetime.now(pytz.UTC) - datetime.timedelta(days=8)
user = UserFactory()
return UserRetirementStatus.objects.create(
user=user,
original_username=user.username,
original_email=user.email,
original_name=user.profile.name,
retired_username=get_retired_username_by_username(user.username),
retired_email=get_retired_email_by_email(user.email),
current_state=state,
last_state=state,
responses="",
created=create_datetime,
modified=create_datetime
)
def _retirement_to_dict(self, retirement, all_fields=False):
"""
Return a dict format of this model to a consistent format for serialization, removing the long text field
`responses` for performance reasons.
"""
retirement_dict = {
u'id': retirement.id,
u'user': {
u'id': retirement.user.id,
u'username': retirement.user.username,
u'email': retirement.user.email,
u'profile': {
u'id': retirement.user.profile.id,
u'name': retirement.user.profile.name
},
},
u'original_username': retirement.original_username,
u'original_email': retirement.original_email,
u'original_name': retirement.original_name,
u'retired_username': retirement.retired_username,
u'retired_email': retirement.retired_email,
u'current_state': {
u'id': retirement.current_state.id,
u'state_name': retirement.current_state.state_name,
u'state_execution_order': retirement.current_state.state_execution_order,
},
u'last_state': {
u'id': retirement.last_state.id,
u'state_name': retirement.last_state.state_name,
u'state_execution_order': retirement.last_state.state_execution_order,
},
u'created': retirement.created,
u'modified': retirement.modified
}
if all_fields:
retirement_dict['responses'] = retirement.responses
return retirement_dict
def _create_users_all_states(self):
return [self._create_retirement(state) for state in RetirementState.objects.all()]
def _get_non_dead_end_states(self):
return [state for state in RetirementState.objects.filter(is_dead_end_state=False)]
def _get_dead_end_states(self):
return [state for state in RetirementState.objects.filter(is_dead_end_state=True)]
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestDeactivateLogout(RetirementTestCase):
"""
Tests the account deactivation/logout endpoint.
"""
def setUp(self):
super(TestDeactivateLogout, self).setUp()
self.test_password = 'password'
self.test_user = UserFactory(password=self.test_password)
UserSocialAuth.objects.create(
user=self.test_user,
provider='some_provider_name',
uid='xyz@gmail.com'
)
UserSocialAuth.objects.create(
user=self.test_user,
provider='some_other_provider_name',
uid='xyz@gmail.com'
)
Registration().register(self.test_user)
self.url = reverse('deactivate_logout')
def build_post(self, password):
return {'password': password}
@mock.patch('openedx.core.djangolib.oauth2_retirement_utils')
def test_user_can_deactivate_self(self, retirement_utils_mock):
"""
Verify a user calling the deactivation endpoint logs out the user, deletes all their SSO tokens,
and creates a user retirement row.
"""
self.client.login(username=self.test_user.username, password=self.test_password)
headers = self.build_jwt_headers(self.test_user)
response = self.client.post(self.url, self.build_post(self.test_password), **headers)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
# make sure the user model is as expected
updated_user = User.objects.get(id=self.test_user.id)
self.assertEqual(get_retired_email_by_email(self.test_user.email), updated_user.email)
self.assertFalse(updated_user.has_usable_password())
self.assertEqual(list(UserSocialAuth.objects.filter(user=self.test_user)), [])
self.assertEqual(list(Registration.objects.filter(user=self.test_user)), [])
self.assertEqual(len(UserRetirementStatus.objects.filter(user_id=self.test_user.id)), 1)
# these retirement utils are tested elsewhere; just make sure we called them
retirement_utils_mock.retire_dop_oauth2_models.assertCalledWith(self.test_user)
retirement_utils_mock.retire_dot_oauth2_models.assertCalledWith(self.test_user)
# make sure the user cannot log in
self.assertFalse(self.client.login(username=self.test_user.username, password=self.test_password))
# make sure that an email has been sent
self.assertEqual(len(mail.outbox), 1)
# ensure that it's been sent to the correct email address
self.assertIn(self.test_user.email, mail.outbox[0].to)
def test_password_mismatch(self):
"""
Verify that the user submitting a mismatched password results in
a rejection.
"""
self.client.login(username=self.test_user.username, password=self.test_password)
headers = self.build_jwt_headers(self.test_user)
response = self.client.post(self.url, self.build_post(self.test_password + "xxxx"), **headers)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_called_twice(self):
"""
Verify a user calling the deactivation endpoint a second time results in a "forbidden"
error, as the user will be logged out.
"""
self.client.login(username=self.test_user.username, password=self.test_password)
headers = self.build_jwt_headers(self.test_user)
response = self.client.post(self.url, self.build_post(self.test_password), **headers)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.client.login(username=self.test_user.username, password=self.test_password)
headers = self.build_jwt_headers(self.test_user)
response = self.client.post(self.url, self.build_post(self.test_password), **headers)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestAccountRetireMailings(RetirementTestCase):
"""
Tests the account retire mailings endpoint.
"""
def setUp(self):
super(TestAccountRetireMailings, self).setUp()
self.test_superuser = SuperuserFactory()
self.test_service_user = UserFactory()
# Should be created in parent setUpClass
retiring_email_lists = RetirementState.objects.get(state_name='RETIRING_EMAIL_LISTS')
self.retirement = self._create_retirement(retiring_email_lists)
self.test_user = self.retirement.user
UserOrgTag.objects.create(user=self.test_user, key='email-optin', org="foo", value="True")
UserOrgTag.objects.create(user=self.test_user, key='email-optin', org="bar", value="True")
self.url = reverse('accounts_retire_mailings')
def build_jwt_headers(self, user):
"""
Helper function for creating headers for the JWT authentication.
"""
token = JwtBuilder(user).build_token([])
headers = {
'HTTP_AUTHORIZATION': 'JWT ' + token
}
return headers
def build_post(self, user):
return {'username': user.username}
def assert_status_and_tag_count(self, headers, expected_status=status.HTTP_204_NO_CONTENT, expected_tag_count=2,
expected_tag_value="False", expected_content=None):
"""
Helper function for making a request to the retire subscriptions endpoint, and asserting the status.
"""
response = self.client.post(self.url, self.build_post(self.test_user), **headers)
self.assertEqual(response.status_code, expected_status)
# Check that the expected number of tags with the correct value exist
tag_count = UserOrgTag.objects.filter(user=self.test_user, value=expected_tag_value).count()
self.assertEqual(tag_count, expected_tag_count)
if expected_content:
self.assertEqual(response.content.strip('"'), expected_content)
def test_superuser_retires_user_subscriptions(self):
"""
Verify a user's subscriptions are retired when a superuser posts to the retire subscriptions endpoint.
"""
headers = self.build_jwt_headers(self.test_superuser)
self.assert_status_and_tag_count(headers)
def test_superuser_retires_user_subscriptions_no_orgtags(self):
"""
Verify the call succeeds when the user doesn't have any org tags.
"""
UserOrgTag.objects.all().delete()
headers = self.build_jwt_headers(self.test_superuser)
self.assert_status_and_tag_count(headers, expected_tag_count=0)
def test_unauthorized_rejection(self):
"""
Verify unauthorized users cannot retire subscriptions.
"""
headers = self.build_jwt_headers(self.test_user)
# User should still have 2 "True" subscriptions.
self.assert_status_and_tag_count(headers, expected_status=status.HTTP_403_FORBIDDEN, expected_tag_value="True")
def test_signal_failure(self):
"""
Verify that if a signal fails the transaction is rolled back and a proper error message is returned.
"""
headers = self.build_jwt_headers(self.test_superuser)
mock_handler = mock.MagicMock()
mock_handler.side_effect = Exception("Tango")
try:
USER_RETIRE_MAILINGS.connect(mock_handler)
# User should still have 2 "True" subscriptions.
self.assert_status_and_tag_count(
headers,
expected_status=status.HTTP_500_INTERNAL_SERVER_ERROR,
expected_tag_value="True",
expected_content="Tango"
)
finally:
USER_RETIRE_MAILINGS.disconnect(mock_handler)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestAccountRetirementList(RetirementTestCase):
"""
Tests the account retirement endpoint.
"""
def setUp(self):
super(TestAccountRetirementList, self).setUp()
self.test_superuser = SuperuserFactory()
self.headers = self.build_jwt_headers(self.test_superuser)
self.url = reverse('accounts_retirement_queue')
self.maxDiff = None
def assert_status_and_user_list(
self,
expected_data,
expected_status=status.HTTP_200_OK,
states_to_request=None,
cool_off_days=7
):
"""
Helper function for making a request to the retire subscriptions endpoint, asserting the status, and
optionally asserting data returned.
"""
if states_to_request is None:
# These are just a couple of random states that should be used in any implementation
states_to_request = ['PENDING', 'LOCKING_ACCOUNT']
else:
# Can pass in RetirementState objects or strings here
try:
states_to_request = [s.state_name for s in states_to_request]
except AttributeError:
states_to_request = states_to_request
data = {'cool_off_days': cool_off_days, 'states': states_to_request}
response = self.client.get(self.url, data, **self.headers)
self.assertEqual(response.status_code, expected_status)
response_data = response.json()
if expected_data:
# These datetimes won't match up due to serialization, but they're inherited fields tested elsewhere
for data in (response_data, expected_data):
for retirement in data:
del retirement['created']
del retirement['modified']
self.assertItemsEqual(response_data, expected_data)
def test_empty(self):
"""
Verify that an empty array is returned if no users are awaiting retirement
"""
self.assert_status_and_user_list([])
def test_users_exist_none_in_correct_status(self):
"""
Verify that users in dead end states are not returned
"""
for state in self._get_dead_end_states():
self._create_retirement(state)
self.assert_status_and_user_list([], states_to_request=self._get_non_dead_end_states())
def test_users_retrieved_in_multiple_states(self):
"""
Verify that if multiple states are requested, learners in each state are returned.
"""
multiple_states = ['PENDING', 'FORUMS_COMPLETE']
for state in multiple_states:
self._create_retirement(RetirementState.objects.get(state_name=state))
data = {'cool_off_days': 0, 'states': multiple_states}
response = self.client.get(self.url, data, **self.headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.json()), 2)
def test_users_exist(self):
"""
Verify users in different states are returned with correct data or filtered out
"""
self.maxDiff = None
retirement_values = []
states_to_request = []
dead_end_states = self._get_dead_end_states()
for retirement in self._create_users_all_states():
if retirement.current_state not in dead_end_states:
states_to_request.append(retirement.current_state)
retirement_values.append(self._retirement_to_dict(retirement))
self.assert_status_and_user_list(retirement_values, states_to_request=self._get_non_dead_end_states())
def test_date_filter(self):
"""
Verifies the functionality of the `cool_off_days` parameter by creating 1 retirement per day for
10 days. Then requests different 1-10 `cool_off_days` to confirm the correct retirements are returned.
"""
retirements = []
days_back_to_test = 10
# Create a retirement per day for the last 10 days, from oldest date to newest. We want these all created
# before we start checking, thus the two loops.
# retirements = [2018-04-10..., 2018-04-09..., 2018-04-08...]
pending_state = RetirementState.objects.get(state_name='PENDING')
for days_back in range(1, days_back_to_test, -1):
create_datetime = datetime.datetime.now(pytz.UTC) - datetime.timedelta(days=days_back)
retirements.append(self._create_retirement(state=pending_state, create_datetime=create_datetime))
# Confirm we get the correct number and data back for each day we add to cool off days
# For each day we add to `cool_off_days` we expect to get one fewer retirement.
for cool_off_days in range(1, days_back_to_test):
# Start with 9 days back
req_days_back = days_back_to_test - cool_off_days
retirement_dicts = [self._retirement_to_dict(ret) for ret in retirements[:cool_off_days]]
self.assert_status_and_user_list(
retirement_dicts,
cool_off_days=req_days_back
)
def test_bad_cool_off_days(self):
"""
Check some bad inputs to make sure we get back the expected status
"""
self.assert_status_and_user_list(None, expected_status=status.HTTP_400_BAD_REQUEST, cool_off_days=-1)
self.assert_status_and_user_list(None, expected_status=status.HTTP_400_BAD_REQUEST, cool_off_days='ABCDERTP')
def test_bad_states(self):
"""
Check some bad inputs to make sure we get back the expected status
"""
self.assert_status_and_user_list(
None,
expected_status=status.HTTP_400_BAD_REQUEST,
states_to_request=['TUNA', 'TACO'])
self.assert_status_and_user_list(None, expected_status=status.HTTP_400_BAD_REQUEST, states_to_request=[])
def test_missing_params(self):
"""
All params are required, make sure that is enforced
"""
response = self.client.get(self.url, **self.headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
response = self.client.get(self.url, {}, **self.headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
response = self.client.get(self.url, {'cool_off_days': 7}, **self.headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
RetirementState.objects.get(state_name='PENDING')
response = self.client.get(self.url, {'states': ['PENDING']}, **self.headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestAccountRetirementRetrieve(RetirementTestCase):
"""
Tests the account retirement retrieval endpoint.
"""
def setUp(self):
super(TestAccountRetirementRetrieve, self).setUp()
self.test_user = UserFactory()
self.test_superuser = SuperuserFactory()
self.url = reverse('accounts_retirement_retrieve', kwargs={'username': self.test_user.username})
self.headers = self.build_jwt_headers(self.test_superuser)
self.maxDiff = None
def assert_status_and_user_data(self, expected_data, expected_status=status.HTTP_200_OK, username_to_find=None):
"""
Helper function for making a request to the retire subscriptions endpoint, asserting the status,
and optionally asserting the expected data.
"""
if username_to_find is not None:
self.url = reverse('accounts_retirement_retrieve', kwargs={'username': username_to_find})
response = self.client.get(self.url, **self.headers)
self.assertEqual(response.status_code, expected_status)
if expected_data is not None:
response_data = response.json()
# These won't match up due to serialization, but they're inherited fields tested elsewhere
for data in (expected_data, response_data):
del data['created']
del data['modified']
self.assertDictEqual(response_data, expected_data)
return response_data
def test_no_retirement(self):
"""
Confirm we get a 404 if a retirement for the user can be found
"""
self.assert_status_and_user_data(None, status.HTTP_404_NOT_FOUND)
def test_retirements_all_states(self):
"""
Create a bunch of retirements and confirm we get back the correct data for each
"""
retirements = []
for state in RetirementState.objects.all():
retirements.append(self._create_retirement(state))
for retirement in retirements:
values = self._retirement_to_dict(retirement)
self.assert_status_and_user_data(values, username_to_find=values['user']['username'])
def test_retrieve_by_old_username(self):
"""
Simulate retrieving a retirement by the old username, after the name has been changed to the hashed one
"""
pending_state = RetirementState.objects.get(state_name='PENDING')
retirement = self._create_retirement(pending_state)
original_username = retirement.user.username
hashed_username = get_retired_username_by_username(original_username)
retirement.user.username = hashed_username
retirement.user.save()
values = self._retirement_to_dict(retirement)
self.assert_status_and_user_data(values, username_to_find=original_username)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestAccountRetirementUpdate(RetirementTestCase):
"""
Tests the account retirement endpoint.
"""
def setUp(self):
super(TestAccountRetirementUpdate, self).setUp()
self.pending_state = RetirementState.objects.get(state_name='PENDING')
self.locking_state = RetirementState.objects.get(state_name='LOCKING_ACCOUNT')
self.retirement = self._create_retirement(self.pending_state)
self.test_user = self.retirement.user
self.test_superuser = SuperuserFactory()
self.headers = self.build_jwt_headers(self.test_superuser)
self.headers['content_type'] = "application/json"
self.url = reverse('accounts_retirement_update')
def update_and_assert_status(self, data, expected_status=status.HTTP_204_NO_CONTENT):
"""
Helper function for making a request to the retire subscriptions endpoint, and asserting the status.
"""
if 'username' not in data:
data['username'] = self.test_user.username
response = self.client.patch(self.url, json.dumps(data), **self.headers)
self.assertEqual(response.status_code, expected_status)
def test_single_update(self):
"""
Basic test to confirm changing state works and saves the given response
"""
data = {'new_state': 'LOCKING_ACCOUNT', 'response': 'this should succeed'}
self.update_and_assert_status(data)
# Refresh the retirment object and confirm the messages and state are correct
retirement = UserRetirementStatus.objects.get(id=self.retirement.id)
self.assertEqual(retirement.current_state, RetirementState.objects.get(state_name='LOCKING_ACCOUNT'))
self.assertEqual(retirement.last_state, RetirementState.objects.get(state_name='PENDING'))
self.assertIn('this should succeed', retirement.responses)
def test_move_through_process(self):
"""
Simulate moving a retirement through the process and confirm they end up in the
correct state, with all relevant response messages logged.
"""
fake_retire_process = [
{'new_state': 'LOCKING_ACCOUNT', 'response': 'accountlockstart'},
{'new_state': 'LOCKING_COMPLETE', 'response': 'accountlockcomplete'},
{'new_state': 'RETIRING_CREDENTIALS', 'response': 'retiringcredentials'},
{'new_state': 'CREDENTIALS_COMPLETE', 'response': 'credentialsretired'},
{'new_state': 'COMPLETE', 'response': 'accountretirementcomplete'},
]
for update_data in fake_retire_process:
self.update_and_assert_status(update_data)
# Refresh the retirment object and confirm the messages and state are correct
retirement = UserRetirementStatus.objects.get(id=self.retirement.id)
self.assertEqual(retirement.current_state, RetirementState.objects.get(state_name='COMPLETE'))
self.assertEqual(retirement.last_state, RetirementState.objects.get(state_name='CREDENTIALS_COMPLETE'))
self.assertIn('accountlockstart', retirement.responses)
self.assertIn('accountlockcomplete', retirement.responses)
self.assertIn('retiringcredentials', retirement.responses)
self.assertIn('credentialsretired', retirement.responses)
self.assertIn('accountretirementcomplete', retirement.responses)
def test_unknown_state(self):
"""
Test that trying to set to an unknown state fails with a 400
"""
data = {'new_state': 'BOGUS_STATE', 'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
def test_bad_vars(self):
"""
Test various ways of sending the wrong variables to make sure they all fail correctly
"""
# No `new_state`
data = {'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
# No `response`
data = {'new_state': 'COMPLETE'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
# Unknown `new_state`
data = {'new_state': 'BOGUS_STATE', 'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
# No `new_state` or `response`
data = {}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
# Unexpected param `should_not_exist`
data = {'should_not_exist': 'bad', 'new_state': 'COMPLETE', 'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
def test_no_retirement(self):
"""
Confirm that trying to operate on a non-existent retirement for an existing user 404s
"""
# Delete the only retirement, created in setUp
UserRetirementStatus.objects.all().delete()
data = {'new_state': 'LOCKING_ACCOUNT', 'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_404_NOT_FOUND)
def test_no_user(self):
"""
Confirm that trying to operate on a non-existent user 404s
"""
data = {'new_state': 'LOCKING_ACCOUNT', 'response': 'this should fail', 'username': 'does not exist'}
self.update_and_assert_status(data, status.HTTP_404_NOT_FOUND)
def test_move_from_dead_end(self):
"""
Confirm that trying to move from a dead end state to any other state fails
"""
retirement = UserRetirementStatus.objects.get(id=self.retirement.id)
retirement.current_state = RetirementState.objects.filter(is_dead_end_state=True)[0]
retirement.save()
data = {'new_state': 'LOCKING_ACCOUNT', 'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
def test_move_backward(self):
"""
Confirm that trying to move to an earlier step in the process fails
"""
retirement = UserRetirementStatus.objects.get(id=self.retirement.id)
retirement.current_state = RetirementState.objects.get(state_name='COMPLETE')
retirement.save()
data = {'new_state': 'PENDING', 'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
def test_move_same(self):
"""
Confirm that trying to move to the same step in the process fails
"""
# Should already be in 'PENDING'
data = {'new_state': 'PENDING', 'response': 'this should fail'}
self.update_and_assert_status(data, status.HTTP_400_BAD_REQUEST)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestAccountRetirementPost(RetirementTestCase):
"""
Tests the account retirement endpoint.
"""
def setUp(self):
super(TestAccountRetirementPost, self).setUp()
self.test_user = UserFactory()
self.test_superuser = SuperuserFactory()
self.original_username = self.test_user.username
self.original_email = self.test_user.email
self.retired_username = get_retired_username_by_username(self.original_username)
self.retired_email = get_retired_email_by_email(self.original_email)
retirement_state = RetirementState.objects.get(state_name='RETIRING_LMS')
self.retirement_status = UserRetirementStatus.create_retirement(self.test_user)
self.retirement_status.current_state = retirement_state
self.retirement_status.last_state = retirement_state
self.retirement_status.save()
SocialLink.objects.create(
user_profile=self.test_user.profile,
platform='Facebook',
social_link='www.facebook.com'
).save()
self.cache_key = UserProfile.country_cache_key_name(self.test_user.id)
cache.set(self.cache_key, 'Timor-leste')
# Enterprise model setup
self.course_id = 'course-v1:edX+DemoX.1+2T2017'
self.enterprise_customer = EnterpriseCustomer.objects.create(
name='test_enterprise_customer',
site=SiteFactory.create()
)
self.enterprise_user = EnterpriseCustomerUser.objects.create(
enterprise_customer=self.enterprise_customer,
user_id=self.test_user.id,
)
self.enterprise_enrollment = EnterpriseCourseEnrollment.objects.create(
enterprise_customer_user=self.enterprise_user,
course_id=self.course_id
)
self.pending_enterprise_user = PendingEnterpriseCustomerUser.objects.create(
enterprise_customer_id=self.enterprise_user.enterprise_customer_id,
user_email=self.test_user.email
)
self.sapsf_audit = SapSuccessFactorsLearnerDataTransmissionAudit.objects.create(
sapsf_user_id=self.test_user.id,
enterprise_course_enrollment_id=self.enterprise_enrollment.id,
completed_timestamp=1,
)
self.consent = DataSharingConsent.objects.create(
username=self.test_user.username,
enterprise_customer=self.enterprise_customer,
)
# Entitlement model setup
self.entitlement = CourseEntitlementFactory.create(user=self.test_user)
self.entitlement_support_detail = CourseEntitlementSupportDetail.objects.create(
entitlement=self.entitlement,
support_user=UserFactory(),
comments='A comment containing potential PII.'
)
# Misc. setup
self.photo_verification = SoftwareSecurePhotoVerificationFactory.create(user=self.test_user)
PendingEmailChangeFactory.create(user=self.test_user)
UserOrgTagFactory.create(user=self.test_user, key='foo', value='bar')
UserOrgTagFactory.create(user=self.test_user, key='cat', value='dog')
CourseEnrollmentAllowedFactory.create(email=self.original_email)
self.course_key = CourseKey.from_string('course-v1:edX+DemoX+Demo_Course')
self.cohort = CourseUserGroup.objects.create(
name="TestCohort",
course_id=self.course_key,
group_type=CourseUserGroup.COHORT
)
self.cohort_assignment = UnregisteredLearnerCohortAssignments.objects.create(
course_user_group=self.cohort,
course_id=self.course_key,
email=self.original_email
)
# setup for doing POST from test client
self.headers = self.build_jwt_headers(self.test_superuser)
self.headers['content_type'] = "application/json"
self.url = reverse('accounts_retire')
def post_and_assert_status(self, data, expected_status=status.HTTP_204_NO_CONTENT):
"""
Helper function for making a request to the retire subscriptions endpoint, and asserting the status.
"""
response = self.client.post(self.url, json.dumps(data), **self.headers)
self.assertEqual(response.status_code, expected_status)
return response
def test_user_profile_pii_has_expected_values(self):
expected_user_profile_pii = {
'name': '',
'meta': '',
'location': '',
'year_of_birth': None,
'gender': None,
'mailing_address': None,
'city': None,
'country': None,
'bio': None,
}
self.assertEqual(expected_user_profile_pii, USER_PROFILE_PII)
def test_retire_user_where_user_does_not_exist(self):
path = 'openedx.core.djangoapps.user_api.accounts.views.is_username_retired'
with mock.patch(path, return_value=False) as mock_retired_username:
data = {'username': 'not_a_user'}
response = self.post_and_assert_status(data, status.HTTP_404_NOT_FOUND)
self.assertFalse(response.content)
mock_retired_username.assert_called_once_with('not_a_user')
def test_retire_user_server_error_is_raised(self):
path = 'openedx.core.djangoapps.user_api.models.UserRetirementStatus.get_retirement_for_retirement_action'
with mock.patch(path, side_effect=Exception('Unexpected Exception')) as mock_get_retirement:
data = {'username': self.test_user.username}
response = self.post_and_assert_status(data, status.HTTP_500_INTERNAL_SERVER_ERROR)
self.assertEqual('Unexpected Exception', text_type(response.json()))
mock_get_retirement.assert_called_once_with(self.original_username)
def test_retire_user_where_user_already_retired(self):
path = 'openedx.core.djangoapps.user_api.accounts.views.is_username_retired'
with mock.patch(path, return_value=True) as mock_is_username_retired:
data = {'username': self.test_user.username}
response = self.post_and_assert_status(data, status.HTTP_404_NOT_FOUND)
self.assertFalse(response.content)
mock_is_username_retired.assert_called_once_with(self.original_username)
def test_retire_user_where_username_not_provided(self):
response = self.post_and_assert_status({}, status.HTTP_404_NOT_FOUND)
expected_response_message = {'message': text_type('The user was not specified.')}
self.assertEqual(expected_response_message, response.json())
@mock.patch('openedx.core.djangoapps.user_api.accounts.views.get_profile_image_names')
@mock.patch('openedx.core.djangoapps.user_api.accounts.views.remove_profile_images')
def test_retire_user(self, mock_remove_profile_images, mock_get_profile_image_names):
data = {'username': self.original_username}
self.post_and_assert_status(data)
self.test_user.refresh_from_db()
self.test_user.profile.refresh_from_db() # pylint: disable=no-member
expected_user_values = {
'first_name': '',
'last_name': '',
'is_active': False,
'username': self.retired_username,
}
for field, expected_value in expected_user_values.iteritems():
self.assertEqual(expected_value, getattr(self.test_user, field))
for field, expected_value in USER_PROFILE_PII.iteritems():
self.assertEqual(expected_value, getattr(self.test_user.profile, field))
self.assertIsNone(self.test_user.profile.profile_image_uploaded_at)
mock_get_profile_image_names.assert_called_once_with(self.original_username)
mock_remove_profile_images.assert_called_once_with(
mock_get_profile_image_names.return_value
)
self.assertFalse(
SocialLink.objects.filter(user_profile=self.test_user.profile).exists()
)
self.assertIsNone(cache.get(self.cache_key))
self._data_sharing_consent_assertions()
self._sapsf_audit_assertions()
self._pending_enterprise_customer_user_assertions()
self._entitlement_support_detail_assertions()
self._photo_verification_assertions()
self.assertFalse(PendingEmailChange.objects.filter(user=self.test_user).exists())
self.assertFalse(UserOrgTag.objects.filter(user=self.test_user).exists())
self.assertFalse(CourseEnrollmentAllowed.objects.filter(email=self.original_email).exists())
self.assertFalse(UnregisteredLearnerCohortAssignments.objects.filter(email=self.original_email).exists())
def test_deletes_pii_from_user_profile(self):
for model_field, value_to_assign in USER_PROFILE_PII.iteritems():
if value_to_assign == '':
value = 'foo'
else:
value = mock.Mock()
setattr(self.test_user.profile, model_field, value)
AccountRetirementView.clear_pii_from_userprofile(self.test_user)
for model_field, value_to_assign in USER_PROFILE_PII.iteritems():
self.assertEqual(value_to_assign, getattr(self.test_user.profile, model_field))
social_links = SocialLink.objects.filter(
user_profile=self.test_user.profile
)
self.assertFalse(social_links.exists())
@mock.patch('openedx.core.djangoapps.user_api.accounts.views.get_profile_image_names')
@mock.patch('openedx.core.djangoapps.user_api.accounts.views.remove_profile_images')
def test_removes_user_profile_images(
self, mock_remove_profile_images, mock_get_profile_image_names
):
test_datetime = datetime.datetime(2018, 1, 1)
self.test_user.profile.profile_image_uploaded_at = test_datetime
AccountRetirementView.delete_users_profile_images(self.test_user)
self.test_user.profile.refresh_from_db() # pylint: disable=no-member
self.assertIsNone(self.test_user.profile.profile_image_uploaded_at)
mock_get_profile_image_names.assert_called_once_with(self.test_user.username)
mock_remove_profile_images.assert_called_once_with(
mock_get_profile_image_names.return_value
)
def test_can_delete_user_profiles_country_cache(self):
AccountRetirementView.delete_users_country_cache(self.test_user)
self.assertIsNone(cache.get(self.cache_key))
def test_can_retire_users_datasharingconsent(self):
AccountRetirementView.retire_users_data_sharing_consent(self.test_user.username, self.retired_username)
self._data_sharing_consent_assertions()
def _data_sharing_consent_assertions(self):
"""
Helper method for asserting that ``DataSharingConsent`` objects are retired.
"""
self.consent.refresh_from_db()
self.assertEqual(self.retired_username, self.consent.username)
test_users_data_sharing_consent = DataSharingConsent.objects.filter(
username=self.original_username
)
self.assertFalse(test_users_data_sharing_consent.exists())
def test_can_retire_users_sap_success_factors_audits(self):
AccountRetirementView.retire_sapsf_data_transmission(self.test_user)
self._sapsf_audit_assertions()
def _sapsf_audit_assertions(self):
"""
Helper method for asserting that ``SapSuccessFactorsLearnerDataTransmissionAudit`` objects are retired.
"""
self.sapsf_audit.refresh_from_db()
self.assertEqual('', self.sapsf_audit.sapsf_user_id)
audits_for_original_user_id = SapSuccessFactorsLearnerDataTransmissionAudit.objects.filter(
sapsf_user_id=self.test_user.id,
)
self.assertFalse(audits_for_original_user_id.exists())
def test_can_retire_user_from_pendingenterprisecustomeruser(self):
AccountRetirementView.retire_user_from_pending_enterprise_customer_user(self.test_user, self.retired_email)
self._pending_enterprise_customer_user_assertions()
def _pending_enterprise_customer_user_assertions(self):
"""
Helper method for asserting that ``PendingEnterpriseCustomerUser`` objects are retired.
"""
self.pending_enterprise_user.refresh_from_db()
self.assertEqual(self.retired_email, self.pending_enterprise_user.user_email)
pending_enterprise_users = PendingEnterpriseCustomerUser.objects.filter(
user_email=self.original_email
)
self.assertFalse(pending_enterprise_users.exists())
def test_course_entitlement_support_detail_comments_are_retired(self):
AccountRetirementView.retire_entitlement_support_detail(self.test_user)
self._entitlement_support_detail_assertions()
def _entitlement_support_detail_assertions(self):
"""
Helper method for asserting that ``CourseEntitleSupportDetail`` objects are retired.
"""
self.entitlement_support_detail.refresh_from_db()
self.assertEqual('', self.entitlement_support_detail.comments)
def _photo_verification_assertions(self):
"""
Helper method for asserting that ``SoftwareSecurePhotoVerification`` objects are retired.
"""
self.photo_verification.refresh_from_db()
self.assertEqual(self.test_user, self.photo_verification.user)
for field in ('name', 'face_image_url', 'photo_id_image_url', 'photo_id_key'):
self.assertEqual('', getattr(self.photo_verification, field))
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Account APIs are only supported in LMS')
class TestLMSAccountRetirementPost(RetirementTestCase, ModuleStoreTestCase):
"""
Tests the LMS account retirement (GDPR P2) endpoint.
"""
def setUp(self):
super(TestLMSAccountRetirementPost, self).setUp()
self.pii_standin = 'PII here'
self.course = CourseFactory()
self.test_user = UserFactory()
self.test_superuser = SuperuserFactory()
self.original_username = self.test_user.username
self.original_email = self.test_user.email
self.retired_username = get_retired_username_by_username(self.original_username)
self.retired_email = get_retired_email_by_email(self.original_email)
retirement_state = RetirementState.objects.get(state_name='RETIRING_LMS')
self.retirement_status = UserRetirementStatus.create_retirement(self.test_user)
self.retirement_status.current_state = retirement_state
self.retirement_status.last_state = retirement_state
self.retirement_status.save()
# wiki data setup
rp = RevisionPlugin.objects.create(article_id=0)
RevisionPluginRevision.objects.create(
revision_number=1,
ip_address="ipaddresss",
plugin=rp,
user=self.test_user,
)
article = Article.objects.create()
ArticleRevision.objects.create(ip_address="ipaddresss", user=self.test_user, article=article)
# ManualEnrollmentAudit setup
course_enrollment = CourseEnrollment.enroll(user=self.test_user, course_key=self.course.id)
ManualEnrollmentAudit.objects.create(
enrollment=course_enrollment, reason=self.pii_standin, enrolled_email=self.pii_standin
)
# CreditRequest and CreditRequirementStatus setup
provider = CreditProvider.objects.create(provider_id="Hogwarts")
credit_course = CreditCourse.objects.create(course_key=self.course.id)
CreditRequest.objects.create(
username=self.test_user.username,
course=credit_course,
provider_id=provider.id,
parameters={self.pii_standin},
)
req = CreditRequirement.objects.create(course_id=credit_course.id)
CreditRequirementStatus.objects.create(username=self.test_user.username, requirement=req)
# ApiAccessRequest setup
site = Site.objects.create()
ApiAccessRequest.objects.create(
user=self.test_user,
site=site,
website=self.pii_standin,
company_address=self.pii_standin,
company_name=self.pii_standin,
reason=self.pii_standin,
)
# SurveyAnswer setup
SurveyAnswer.objects.create(user=self.test_user, field_value=self.pii_standin, form_id=0)
# other setup
PendingNameChange.objects.create(user=self.test_user, new_name=self.pii_standin, rationale=self.pii_standin)
PasswordHistory.objects.create(user=self.test_user, password=self.pii_standin)
# setup for doing POST from test client
self.headers = self.build_jwt_headers(self.test_superuser)
self.headers['content_type'] = "application/json"
self.url = reverse('accounts_retire_misc')
def post_and_assert_status(self, data, expected_status=status.HTTP_204_NO_CONTENT):
"""
Helper function for making a request to the retire subscriptions endpoint, and asserting the status.
"""
response = self.client.post(self.url, json.dumps(data), **self.headers)
self.assertEqual(response.status_code, expected_status)
return response
def test_retire_user(self):
# check that rows that will not exist after retirement exist now
self.assertTrue(CreditRequest.objects.filter(username=self.test_user.username).exists())
self.assertTrue(CreditRequirementStatus.objects.filter(username=self.test_user.username).exists())
self.assertTrue(PendingNameChange.objects.filter(user=self.test_user).exists())
retirement = UserRetirementStatus.get_retirement_for_retirement_action(self.test_user.username)
data = {'username': self.original_username}
self.post_and_assert_status(data)
self.test_user.refresh_from_db()
self.test_user.profile.refresh_from_db() # pylint: disable=no-member
self.assertEqual(RevisionPluginRevision.objects.get(user=self.test_user).ip_address, None)
self.assertEqual(ArticleRevision.objects.get(user=self.test_user).ip_address, None)
self.assertFalse(PendingNameChange.objects.filter(user=self.test_user).exists())
self.assertEqual(PasswordHistory.objects.get(user=self.test_user).password, '')
self.assertEqual(
ManualEnrollmentAudit.objects.get(
enrollment=CourseEnrollment.objects.get(user=self.test_user)
).enrolled_email,
retirement.retired_email
)
self.assertFalse(CreditRequest.objects.filter(username=self.test_user.username).exists())
self.assertTrue(CreditRequest.objects.filter(username=retirement.retired_username).exists())
self.assertEqual(CreditRequest.objects.get(username=retirement.retired_username).parameters, {})
self.assertFalse(CreditRequirementStatus.objects.filter(username=self.test_user.username).exists())
self.assertTrue(CreditRequirementStatus.objects.filter(username=retirement.retired_username).exists())
self.assertEqual(CreditRequirementStatus.objects.get(username=retirement.retired_username).reason, {})
retired_api_access_request = ApiAccessRequest.objects.get(user=self.test_user)
self.assertEqual(retired_api_access_request.website, '')
self.assertEqual(retired_api_access_request.company_address, '')
self.assertEqual(retired_api_access_request.company_name, '')
self.assertEqual(retired_api_access_request.reason, '')
self.assertEqual(SurveyAnswer.objects.get(user=self.test_user).field_value, '')

91
openedx/core/djangoapps/user_api/accounts/views.py
View File

@@ -66,10 +66,16 @@ from student.models import (
from student.views.login import AuthFailedError, LoginFailures
from ..errors import AccountUpdateError, AccountValidationError, UserNotAuthorized, UserNotFound
from ..models import RetirementState, RetirementStateError, UserOrgTag, UserRetirementStatus
from ..models import (
RetirementState,
RetirementStateError,
UserOrgTag,
UserRetirementPartnerReportingStatus,
UserRetirementStatus
)
from .api import get_account_settings, update_account_settings
from .permissions import CanDeactivateUser, CanRetireUser
from .serializers import UserRetirementStatusSerializer
from .serializers import UserRetirementPartnerReportSerializer, UserRetirementStatusSerializer
from .signals import USER_RETIRE_MAILINGS
from ..message_types import DeletionNotificationMessage
@@ -515,6 +521,87 @@ def _set_unusable_password(user):
user.save()
class AccountRetirementPartnerReportView(ViewSet):
"""
Provides API endpoints for managing partner reporting of retired
users.
"""
authentication_classes = (JwtAuthentication,)
permission_classes = (permissions.IsAuthenticated, CanRetireUser,)
parser_classes = (JSONParser,)
serializer_class = UserRetirementStatusSerializer
def _get_orgs_for_user(self, user):
"""
Returns a set of orgs that the user has enrollments with
"""
orgs = set()
for enrollment in user.courseenrollment_set.all():
org = enrollment.course.org
# Org can concievably be blank or this bogus default value
if org and org != 'outdated_entry':
orgs.add(enrollment.course.org)
return orgs
def retirement_partner_report(self, request): # pylint: disable=unused-argument
"""
POST /api/user/v1/accounts/retirement_partner_report/
Returns the list of UserRetirementPartnerReportingStatus users
that are not already being processed and updates their status
to indicate they are currently being processed.
"""
retirement_statuses = UserRetirementPartnerReportingStatus.objects.filter(
is_being_processed=False
).order_by('id')
retirements = [
{
'original_username': retirement.original_username,
'original_email': retirement.original_email,
'original_name': retirement.original_name,
'orgs': self._get_orgs_for_user(retirement.user)
}
for retirement in retirement_statuses
]
serializer = UserRetirementPartnerReportSerializer(retirements, many=True)
retirement_statuses.update(is_being_processed=True)
return Response(serializer.data)
def retirement_partner_cleanup(self, request):
"""
DELETE /api/user/v1/accounts/retirement_partner_report/
[{'original_username': 'user1'}, {'original_username': 'user2'}, ...]
Deletes UserRetirementPartnerReportingStatus objects for a list of users
that have been reported on.
"""
usernames = [u['original_username'] for u in request.data]
if not usernames:
return Response('No original_usernames given.', status=status.HTTP_400_BAD_REQUEST)
retirement_statuses = UserRetirementPartnerReportingStatus.objects.filter(
is_being_processed=True,
original_username__in=usernames
)
if len(usernames) != len(retirement_statuses):
return Response(
'{} original_usernames given, only {} found!'.format(len(usernames), len(retirement_statuses)),
status=status.HTTP_400_BAD_REQUEST
)
retirement_statuses.delete()
return Response(status=status.HTTP_204_NO_CONTENT)
class AccountRetirementStatusView(ViewSet):
"""
Provides API endpoints for managing the user retirement process.

37
openedx/core/djangoapps/user_api/migrations/0004_userretirementpartnerreportingstatus.py Normal file
View File

@@ -0,0 +1,37 @@
# -*- coding: utf-8 -*-
# Generated by Django 1.11.13 on 2018-06-13 20:54
from __future__ import unicode_literals
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
import django.utils.timezone
import model_utils.fields
class Migration(migrations.Migration):
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
('user_api', '0003_userretirementrequest'),
]
operations = [
migrations.CreateModel(
name='UserRetirementPartnerReportingStatus',
fields=[
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('created', model_utils.fields.AutoCreatedField(default=django.utils.timezone.now, editable=False, verbose_name='created')),
('modified', model_utils.fields.AutoLastModifiedField(default=django.utils.timezone.now, editable=False, verbose_name='modified')),
('original_username', models.CharField(db_index=True, max_length=150)),
('original_email', models.EmailField(db_index=True, max_length=254)),
('original_name', models.CharField(blank=True, db_index=True, max_length=255)),
('is_being_processed', models.BooleanField(default=False)),
('user', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
],
options={
'verbose_name': 'User Retirement Reporting Status',
'verbose_name_plural': 'User Retirement Reporting Statuses',
},
),
]

25
openedx/core/djangoapps/user_api/models.py
View File

@@ -167,6 +167,31 @@ class RetirementState(models.Model):
return cls.objects.all().values_list('state_name', flat=True)
class UserRetirementPartnerReportingStatus(TimeStampedModel):
"""
When a user has been retired from LMS it will still need to be reported out to
partners so they can forget the user also. This process happens on a very different,
and asynchronous, timeline than LMS retirement and only impacts a subset of learners
so it maintains a queue. This queue is populated as part of the LMS retirement
process.
"""
user = models.OneToOneField(User)
original_username = models.CharField(max_length=150, db_index=True)
original_email = models.EmailField(db_index=True)
original_name = models.CharField(max_length=255, blank=True, db_index=True)
is_being_processed = models.BooleanField(default=False)
class Meta(object):
verbose_name = 'User Retirement Reporting Status'
verbose_name_plural = 'User Retirement Reporting Statuses'
def __unicode__(self):
return u'UserRetirementPartnerReportingStatus: {} is being processed: {}'.format(
self.user,
self.is_being_processed
)
class UserRetirementRequest(TimeStampedModel):
"""
Records and perists every user retirement request.

2
openedx/core/djangoapps/user_api/tests/test_views.py
View File

@@ -20,7 +20,7 @@ from six import text_type
from social_django.models import UserSocialAuth, Partial
from django_comment_common import models
from openedx.core.djangoapps.user_api.accounts.tests.test_views import RetirementTestCase
from openedx.core.djangoapps.user_api.accounts.tests.test_retirement_views import RetirementTestCase
from openedx.core.djangoapps.user_api.models import UserRetirementStatus
from openedx.core.djangoapps.site_configuration.helpers import get_value
from openedx.core.lib.api.test_utils import ApiTestCase, TEST_API_KEY

11
openedx/core/djangoapps/user_api/urls.py
View File

@@ -9,6 +9,7 @@ from ..profile_images.views import ProfileImageView
from .accounts.views import (
AccountDeactivationView,
AccountRetireMailingsView,
AccountRetirementPartnerReportView,
AccountRetirementStatusView,
AccountRetirementView,
AccountViewSet,
@@ -32,6 +33,11 @@ ACCOUNT_DETAIL = AccountViewSet.as_view({
'patch': 'partial_update',
})
PARTNER_REPORT = AccountRetirementPartnerReportView.as_view({
'post': 'retirement_partner_report',
'delete': 'retirement_partner_cleanup'
})
RETIREMENT_QUEUE = AccountRetirementStatusView.as_view({
'get': 'retirement_queue'
})
@@ -98,6 +104,11 @@ urlpatterns = [
RETIREMENT_RETRIEVE,
name='accounts_retirement_retrieve'
),
url(
r'^v1/accounts/retirement_partner_report/$',
PARTNER_REPORT,
name='accounts_retirement_partner_report'
),
url(
r'^v1/accounts/retirement_queue/$',
RETIREMENT_QUEUE,