Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d1bf43dd2add8d4912ad2e4f8edde72bb76a8bc9
edx-platform/scripts/user_retirement/tests/utils/test_edx_api.py
Deborah Kaplan a7f6cf160c feat: removing the from the retirement pipeline (#35215) 
although the demographics IDA has not been entirely deprecated as yet,
it is about to be, so removing from the retirement pipeline.
2024-08-01 13:23:50 -04:00

545 lines
19 KiB
Python
Raw Blame History

"""
Tests for edX API calls.
"""
import unittest
from unittest.mock import DEFAULT, patch
from urllib.parse import urljoin
import requests
import responses
from ddt import data, ddt, unpack
from requests.exceptions import ConnectionError, HTTPError
from responses import GET, PATCH, POST, matchers
from responses.registries import OrderedRegistry
from scripts.user_retirement.tests.mixins import OAuth2Mixin
from scripts.user_retirement.tests.retirement_helpers import (
FAKE_DATETIME_OBJECT,
FAKE_ORIGINAL_USERNAME,
FAKE_RESPONSE_MESSAGE,
FAKE_USERNAME_MAPPING,
FAKE_USERNAMES,
TEST_RETIREMENT_QUEUE_STATES,
TEST_RETIREMENT_STATE,
get_fake_user_retirement,
)
from scripts.user_retirement.utils import edx_api
class BackoffTriedException(Exception):
"""
Raise this from a backoff handler to indicate that backoff was tried.
"""
@ddt
class TestLmsApi(OAuth2Mixin, unittest.TestCase):
"""
Test the edX LMS API client.
"""
@responses.activate(registry=OrderedRegistry)
def setUp(self):
super().setUp()
self.mock_access_token_response()
self.lms_base_url = 'http://localhost:18000/'
self.lms_api = edx_api.LmsApi(
self.lms_base_url,
self.lms_base_url,
'the_client_id',
'the_client_secret'
)
def tearDown(self):
super().tearDown()
responses.reset()
@patch.object(edx_api.LmsApi, 'learners_to_retire')
def test_learners_to_retire(self, mock_method):
params = {
'states': TEST_RETIREMENT_QUEUE_STATES,
'cool_off_days': 365,
}
responses.add(
GET,
urljoin(self.lms_base_url, 'api/user/v1/accounts/retirement_queue/'),
match=[matchers.query_param_matcher(params)],
)
self.lms_api.learners_to_retire(
TEST_RETIREMENT_QUEUE_STATES, cool_off_days=365)
mock_method.assert_called_once_with(
TEST_RETIREMENT_QUEUE_STATES, cool_off_days=365)
@patch.object(edx_api.LmsApi, 'get_learners_by_date_and_status')
def test_get_learners_by_date_and_status(self, mock_method):
query_params = {
'start_date': FAKE_DATETIME_OBJECT.strftime('%Y-%m-%d'),
'end_date': FAKE_DATETIME_OBJECT.strftime('%Y-%m-%d'),
'state': TEST_RETIREMENT_STATE,
}
responses.add(
GET,
urljoin(self.lms_base_url, 'api/user/v1/accounts/retirements_by_status_and_date/'),
match=[matchers.query_param_matcher(query_params)]
)
self.lms_api.get_learners_by_date_and_status(
state_to_request=TEST_RETIREMENT_STATE,
start_date=FAKE_DATETIME_OBJECT,
end_date=FAKE_DATETIME_OBJECT
)
mock_method.assert_called_once_with(
state_to_request=TEST_RETIREMENT_STATE,
start_date=FAKE_DATETIME_OBJECT,
end_date=FAKE_DATETIME_OBJECT
)
@patch.object(edx_api.LmsApi, 'get_learner_retirement_state')
def test_get_learner_retirement_state(self, mock_method):
responses.add(
GET,
urljoin(self.lms_base_url, f'api/user/v1/accounts/{FAKE_ORIGINAL_USERNAME}/retirement_status/'),
)
self.lms_api.get_learner_retirement_state(
username=FAKE_ORIGINAL_USERNAME
)
mock_method.assert_called_once_with(
username=FAKE_ORIGINAL_USERNAME
)
@patch.object(edx_api.LmsApi, 'update_learner_retirement_state')
def test_update_leaner_retirement_state(self, mock_method):
json_data = {
'username': FAKE_ORIGINAL_USERNAME,
'new_state': TEST_RETIREMENT_STATE,
'response': FAKE_RESPONSE_MESSAGE,
}
responses.add(
PATCH,
urljoin(self.lms_base_url, 'api/user/v1/accounts/update_retirement_status/'),
match=[matchers.json_params_matcher(json_data)]
)
self.lms_api.update_learner_retirement_state(
username=FAKE_ORIGINAL_USERNAME,
new_state_name=TEST_RETIREMENT_STATE,
message=FAKE_RESPONSE_MESSAGE
)
mock_method.assert_called_once_with(
username=FAKE_ORIGINAL_USERNAME,
new_state_name=TEST_RETIREMENT_STATE,
message=FAKE_RESPONSE_MESSAGE
)
@data(
{
'api_url': 'api/user/v1/accounts/deactivate_logout/',
'mock_method': 'retirement_deactivate_logout',
'method': 'POST',
},
{
'api_url': 'api/discussion/v1/accounts/retire_forum/',
'mock_method': 'retirement_retire_forum',
'method': 'POST',
},
{
'api_url': 'api/user/v1/accounts/retire_mailings/',
'mock_method': 'retirement_retire_mailings',
'method': 'POST',
},
{
'api_url': 'api/enrollment/v1/unenroll/',
'mock_method': 'retirement_unenroll',
'method': 'POST',
},
{
'api_url': 'api/edxnotes/v1/retire_user/',
'mock_method': 'retirement_retire_notes',
'method': 'POST',
},
{
'api_url': 'api/user/v1/accounts/retire_misc/',
'mock_method': 'retirement_lms_retire_misc',
'method': 'POST',
},
{
'api_url': 'api/user/v1/accounts/retire/',
'mock_method': 'retirement_lms_retire',
'method': 'POST',
},
{
'api_url': 'api/user/v1/accounts/retirement_partner_report/',
'mock_method': 'retirement_partner_queue',
'method': 'PUT',
},
)
@unpack
@patch.multiple(
'scripts.user_retirement.utils.edx_api.LmsApi',
retirement_deactivate_logout=DEFAULT,
retirement_retire_forum=DEFAULT,
retirement_retire_mailings=DEFAULT,
retirement_unenroll=DEFAULT,
retirement_retire_notes=DEFAULT,
retirement_lms_retire_misc=DEFAULT,
retirement_lms_retire=DEFAULT,
retirement_partner_queue=DEFAULT,
)
def test_learner_retirement(self, api_url, mock_method, method, **kwargs):
json_data = {
'username': FAKE_ORIGINAL_USERNAME,
}
responses.add(
method,
urljoin(self.lms_base_url, api_url),
match=[matchers.json_params_matcher(json_data)]
)
getattr(self.lms_api, mock_method)(get_fake_user_retirement(original_username=FAKE_ORIGINAL_USERNAME))
kwargs[mock_method].assert_called_once_with(get_fake_user_retirement(original_username=FAKE_ORIGINAL_USERNAME))
@patch.object(edx_api.LmsApi, 'retirement_partner_report')
def test_retirement_partner_report(self, mock_method):
responses.add(
POST,
urljoin(self.lms_base_url, 'api/user/v1/accounts/retirement_partner_report/')
)
self.lms_api.retirement_partner_report(
learner=get_fake_user_retirement(
original_username=FAKE_ORIGINAL_USERNAME
)
)
mock_method.assert_called_once_with(
learner=get_fake_user_retirement(
original_username=FAKE_ORIGINAL_USERNAME
)
)
@patch.object(edx_api.LmsApi, 'retirement_partner_cleanup')
def test_retirement_partner_cleanup(self, mock_method):
json_data = FAKE_USERNAMES
responses.add(
POST,
urljoin(self.lms_base_url, 'api/user/v1/accounts/retirement_partner_report_cleanup/'),
match=[matchers.json_params_matcher(json_data)]
)
self.lms_api.retirement_partner_cleanup(
usernames=FAKE_USERNAMES
)
mock_method.assert_called_once_with(
usernames=FAKE_USERNAMES
)
@patch.object(edx_api.LmsApi, 'retirement_retire_proctoring_data')
def test_retirement_retire_proctoring_data(self, mock_method):
learner = get_fake_user_retirement()
responses.add(
POST,
urljoin(self.lms_base_url, f"api/edx_proctoring/v1/retire_user/{learner['user']['id']}/"),
)
self.lms_api.retirement_retire_proctoring_data()
mock_method.assert_called_once()
@patch.object(edx_api.LmsApi, 'retirement_retire_proctoring_backend_data')
def test_retirement_retire_proctoring_backend_data(self, mock_method):
learner = get_fake_user_retirement()
responses.add(
POST,
urljoin(self.lms_base_url, f"api/edx_proctoring/v1/retire_backend_user/{learner['user']['id']}/"),
)
self.lms_api.retirement_retire_proctoring_backend_data()
mock_method.assert_called_once()
@patch.object(edx_api.LmsApi, 'replace_lms_usernames')
def test_replace_lms_usernames(self, mock_method):
json_data = {
'username_mappings': FAKE_USERNAME_MAPPING
}
responses.add(
POST,
urljoin(self.lms_base_url, 'api/user/v1/accounts/replace_usernames/'),
match=[matchers.json_params_matcher(json_data)]
)
self.lms_api.replace_lms_usernames(
username_mappings=FAKE_USERNAME_MAPPING
)
mock_method.assert_called_once_with(
username_mappings=FAKE_USERNAME_MAPPING
)
@patch.object(edx_api.LmsApi, 'replace_forums_usernames')
def test_replace_forums_usernames(self, mock_method):
json_data = {
'username_mappings': FAKE_USERNAME_MAPPING
}
responses.add(
POST,
urljoin(self.lms_base_url, 'api/discussion/v1/accounts/replace_usernames/'),
match=[matchers.json_params_matcher(json_data)]
)
self.lms_api.replace_forums_usernames(
username_mappings=FAKE_USERNAME_MAPPING
)
mock_method.assert_called_once_with(
username_mappings=FAKE_USERNAME_MAPPING
)
@data(504, 500)
@patch('scripts.user_retirement.utils.edx_api._backoff_handler')
@patch.object(edx_api.LmsApi, 'learners_to_retire')
def test_retrieve_learner_queue_backoff(
self,
svr_status_code,
mock_backoff_handler,
mock_learners_to_retire
):
mock_backoff_handler.side_effect = BackoffTriedException
params = {
'states': TEST_RETIREMENT_QUEUE_STATES,
'cool_off_days': 365,
}
response = requests.Response()
response.status_code = svr_status_code
responses.add(
GET,
urljoin(self.lms_base_url, 'api/user/v1/accounts/retirement_queue/'),
status=200,
match=[matchers.query_param_matcher(params)],
)
mock_learners_to_retire.side_effect = HTTPError(response=response)
with self.assertRaises(BackoffTriedException):
self.lms_api.learners_to_retire(
TEST_RETIREMENT_QUEUE_STATES, cool_off_days=365)
@data(104)
@responses.activate
@patch('scripts.user_retirement.utils.edx_api._backoff_handler')
@patch.object(edx_api.LmsApi, 'retirement_partner_cleanup')
def test_retirement_partner_cleanup_backoff_on_connection_error(
self,
svr_status_code,
mock_backoff_handler,
mock_retirement_partner_cleanup
):
mock_backoff_handler.side_effect = BackoffTriedException
response = requests.Response()
response.status_code = svr_status_code
mock_retirement_partner_cleanup.retirement_partner_cleanup.side_effect = ConnectionError(
response=response
)
with self.assertRaises(BackoffTriedException):
self.lms_api.retirement_partner_cleanup([{'original_username': 'test'}])
class TestEcommerceApi(OAuth2Mixin, unittest.TestCase):
"""
Test the edX Ecommerce API client.
"""
@responses.activate(registry=OrderedRegistry)
def setUp(self):
super().setUp()
self.mock_access_token_response()
self.lms_base_url = 'http://localhost:18000/'
self.ecommerce_base_url = 'http://localhost:18130/'
self.ecommerce_api = edx_api.EcommerceApi(
self.lms_base_url,
self.ecommerce_base_url,
'the_client_id',
'the_client_secret'
)
def tearDown(self):
super().tearDown()
responses.reset()
@patch.object(edx_api.EcommerceApi, 'retire_learner')
def test_retirement_partner_report(self, mock_method):
json_data = {
'username': FAKE_ORIGINAL_USERNAME,
}
responses.add(
POST,
urljoin(self.lms_base_url, 'api/v2/user/retire/'),
match=[matchers.json_params_matcher(json_data)]
)
self.ecommerce_api.retire_learner(
learner=get_fake_user_retirement(original_username=FAKE_ORIGINAL_USERNAME)
)
mock_method.assert_called_once_with(
learner=get_fake_user_retirement(original_username=FAKE_ORIGINAL_USERNAME)
)
@patch.object(edx_api.EcommerceApi, 'retire_learner')
def get_tracking_key(self, mock_method):
original_username = {
'original_username': get_fake_user_retirement(original_username=FAKE_ORIGINAL_USERNAME)
}
responses.add(
GET,
urljoin(self.lms_base_url, f"api/v2/retirement/tracking_id/{original_username}/"),
)
self.ecommerce_api.get_tracking_key(
learner=get_fake_user_retirement(original_username=FAKE_ORIGINAL_USERNAME)
)
mock_method.assert_called_once_with(
learner=get_fake_user_retirement(original_username=FAKE_ORIGINAL_USERNAME)
)
@patch.object(edx_api.EcommerceApi, 'replace_usernames')
def test_replace_usernames(self, mock_method):
json_data = {
"username_mappings": FAKE_USERNAME_MAPPING
}
responses.add(
POST,
urljoin(self.lms_base_url, 'api/v2/user_management/replace_usernames/'),
match=[matchers.json_params_matcher(json_data)]
)
self.ecommerce_api.replace_usernames(
username_mappings=FAKE_USERNAME_MAPPING
)
mock_method.assert_called_once_with(
username_mappings=FAKE_USERNAME_MAPPING
)
class TestCredentialApi(OAuth2Mixin, unittest.TestCase):
"""
Test the edX Credential API client.
"""
@responses.activate(registry=OrderedRegistry)
def setUp(self):
super().setUp()
self.mock_access_token_response()
self.lms_base_url = 'http://localhost:18000/'
self.credentials_base_url = 'http://localhost:18150/'
self.credentials_api = edx_api.CredentialsApi(
self.lms_base_url,
self.credentials_base_url,
'the_client_id',
'the_client_secret'
)
def tearDown(self):
super().tearDown()
responses.reset()
@patch.object(edx_api.CredentialsApi, 'retire_learner')
def test_retire_learner(self, mock_method):
json_data = {
'username': FAKE_ORIGINAL_USERNAME
}
responses.add(
POST,
urljoin(self.credentials_base_url, 'user/retire/'),
match=[matchers.json_params_matcher(json_data)]
)
self.credentials_api.retire_learner(
learner=get_fake_user_retirement(original_username=FAKE_ORIGINAL_USERNAME)
)
mock_method.assert_called_once_with(
learner=get_fake_user_retirement(original_username=FAKE_ORIGINAL_USERNAME)
)
@patch.object(edx_api.CredentialsApi, 'replace_usernames')
def test_replace_usernames(self, mock_method):
json_data = {
"username_mappings": FAKE_USERNAME_MAPPING
}
responses.add(
POST,
urljoin(self.credentials_base_url, 'api/v2/replace_usernames/'),
match=[matchers.json_params_matcher(json_data)]
)
self.credentials_api.replace_usernames(
username_mappings=FAKE_USERNAME_MAPPING
)
mock_method.assert_called_once_with(
username_mappings=FAKE_USERNAME_MAPPING
)
class TestDiscoveryApi(OAuth2Mixin, unittest.TestCase):
"""
Test the edX Discovery API client.
"""
@responses.activate(registry=OrderedRegistry)
def setUp(self):
super().setUp()
self.mock_access_token_response()
self.lms_base_url = 'http://localhost:18000/'
self.discovery_base_url = 'http://localhost:18150/'
self.discovery_api = edx_api.DiscoveryApi(
self.lms_base_url,
self.discovery_base_url,
'the_client_id',
'the_client_secret'
)
def tearDown(self):
super().tearDown()
responses.reset()
@patch.object(edx_api.DiscoveryApi, 'replace_usernames')
def test_replace_usernames(self, mock_method):
json_data = {
"username_mappings": FAKE_USERNAME_MAPPING
}
responses.add(
POST,
urljoin(self.discovery_base_url, 'api/v1/replace_usernames/'),
match=[matchers.json_params_matcher(json_data)]
)
self.discovery_api.replace_usernames(
username_mappings=FAKE_USERNAME_MAPPING
)
mock_method.assert_called_once_with(
username_mappings=FAKE_USERNAME_MAPPING
)
class TestLicenseManagerApi(OAuth2Mixin, unittest.TestCase):
"""
Test the edX License Manager API client.
"""
@responses.activate(registry=OrderedRegistry)
def setUp(self):
super().setUp()
self.mock_access_token_response()
self.lms_base_url = 'http://localhost:18000/'
self.license_manager_base_url = 'http://localhost:18170/'
self.license_manager_api = edx_api.LicenseManagerApi(
self.lms_base_url,
self.license_manager_base_url,
'the_client_id',
'the_client_secret'
)
def tearDown(self):
super().tearDown()
responses.reset()
@patch.object(edx_api.LicenseManagerApi, 'retire_learner')
def test_retire_learner(self, mock_method):
json_data = {
'lms_user_id': get_fake_user_retirement()['user']['id'],
'original_username': FAKE_ORIGINAL_USERNAME,
}
responses.add(
POST,
urljoin(self.license_manager_base_url, 'api/v1/retire_user/'),
match=[matchers.json_params_matcher(json_data)]
)
self.license_manager_api.retire_learner(
learner=get_fake_user_retirement(
original_username=FAKE_ORIGINAL_USERNAME
)
)
mock_method.assert_called_once_with(
learner=get_fake_user_retirement(
original_username=FAKE_ORIGINAL_USERNAME
)
)
Reference in New Issue View Git Blame Copy Permalink