Skip to content

E
edx-platform
Commit 70eaf189 by Ahsan Ulhaq
Options

caching for requests to credentials service 

ECOM-3278
parent 51815136
Showing with 381 additions and 92 deletions
cms/djangoapps/contentstore/views/tests/test_programs.py
......@@ -21,8 +21,6 @@ class TestProgramListing(ProgramsApiConfigMixin, ProgramsDataMixin, SharedModule
ClientFactory(name=ProgramsApiConfig.OAUTH2_CLIENT_NAME, client_type=CONFIDENTIAL)
self.create_programs_config()
self.staff = UserFactory(is_staff=True)
self.client.login(username=self.staff.username, password='test')
......@@ -50,6 +48,7 @@ class TestProgramListing(ProgramsApiConfigMixin, ProgramsDataMixin, SharedModule
student = UserFactory(is_staff=False)
self.client.login(username=student.username, password='test')
self.create_programs_config()
self.mock_programs_api()
response = self.client.get(self.studio_home)
......@@ -60,6 +59,7 @@ class TestProgramListing(ProgramsApiConfigMixin, ProgramsDataMixin, SharedModule
"""Verify that the programs tab and creation button can be rendered when config is enabled."""
# When no data is provided, expect creation prompt.
self.create_programs_config()
self.mock_programs_api(data={'results': []})
response = self.client.get(self.studio_home)
......
common/djangoapps/student/views.py
......@@ -2432,7 +2432,7 @@ def _get_xseries_credentials(user):
programs_credentials = get_user_program_credentials(user)
credentials_data = []
for program in programs_credentials:
if program.get('status') == 'active':
if program.get('category') == 'xseries':
try:
program_data = {
'display_name': program['name'],
......
lms/urls.py
......@@ -102,6 +102,8 @@ urlpatterns = (
url(r'^api/commerce/', include('commerce.api.urls', namespace='commerce_api')),
url(r'^api/credit/', include('openedx.core.djangoapps.credit.urls', app_name="credit", namespace='credit')),
url(r'^rss_proxy/', include('rss_proxy.urls', namespace='rss_proxy')),
url(r'^api/organizations/', include(
'openedx.core.djangoapps.organization_api.urls', app_name='organization_api', namespace='organization_api')),
)
if settings.FEATURES["ENABLE_COMBINED_LOGIN_REGISTRATION"]:
......
openedx/core/djangoapps/credentials/migrations/0002_credentialsapiconfig_cache_ttl.py 0 → 100644
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('credentials', '0001_initial'),
]
operations = [
migrations.AddField(
model_name='credentialsapiconfig',
name='cache_ttl',
field=models.PositiveIntegerField(default=0, help_text='Specified in seconds. Enable caching by setting this to a value greater than 0.', verbose_name='Cache Time To Live'),
),
]
openedx/core/djangoapps/credentials/models.py
......@@ -17,6 +17,7 @@ class CredentialsApiConfig(ConfigurationModel):
"""
OAUTH2_CLIENT_NAME = 'credentials'
API_NAME = 'credentials'
CACHE_KEY = 'credentials.api.data'
internal_service_url = models.URLField(verbose_name=_("Internal Service URL"))
public_service_url = models.URLField(verbose_name=_("Public Service URL"))
......@@ -35,6 +36,13 @@ class CredentialsApiConfig(ConfigurationModel):
"Enable authoring of Credential Service credentials in Studio."
)
)
cache_ttl = models.PositiveIntegerField(
verbose_name=_("Cache Time To Live"),
default=0,
help_text=_(
"Specified in seconds. Enable caching by setting this to a value greater than 0."
)
)
def __unicode__(self):
return self.public_api_url
......@@ -67,3 +75,8 @@ class CredentialsApiConfig(ConfigurationModel):
be enabled or not.
"""
return self.enabled and self.enable_studio_authoring
@property
def is_cache_enabled(self):
"""Whether responses from the Credentials API will be cached."""
return self.cache_ttl > 0
openedx/core/djangoapps/credentials/tests/mixins.py
......@@ -15,6 +15,7 @@ class CredentialsApiConfigMixin(object):
'public_service_url': 'http://public.credentials.org/',
'enable_learner_issuance': True,
'enable_studio_authoring': True,
'cache_ttl': 0,
}
def create_credentials_config(self, **kwargs):
......@@ -99,7 +100,7 @@ class CredentialsDataMixin(object):
}
CREDENTIALS_NEXT_API_RESPONSE = {
"next": 'next_page_url',
"next": None,
"results": [
{
"id": 7,
......@@ -140,9 +141,9 @@ class CredentialsDataMixin(object):
body = json.dumps(data)
if is_next_page:
next_page_data = self.CREDENTIALS_NEXT_API_RESPONSE
next_page_body = json.dumps(next_page_data)
next_page_url = internal_api_url + '/user_credentials/?page=2&username=' + user.username
self.CREDENTIALS_NEXT_API_RESPONSE['next'] = next_page_url
next_page_body = json.dumps(self.CREDENTIALS_NEXT_API_RESPONSE)
httpretty.register_uri(
httpretty.GET, next_page_url, body=body, content_type='application/json', status=status_code
)
......
openedx/core/djangoapps/credentials/tests/test_utils.py
"""Tests covering Credentials utilities."""
from django.core.cache import cache
from django.test import TestCase
import httpretty
from oauth2_provider.tests.factories import ClientFactory
......@@ -26,6 +27,8 @@ class TestCredentialsRetrieval(ProgramsApiConfigMixin, CredentialsApiConfigMixin
ClientFactory(name=ProgramsApiConfig.OAUTH2_CLIENT_NAME, client_type=CONFIDENTIAL)
self.user = UserFactory()
cache.clear()
@httpretty.activate
def test_get_user_credentials(self):
"""Verify user credentials data can be retrieve."""
......@@ -35,6 +38,30 @@ class TestCredentialsRetrieval(ProgramsApiConfigMixin, CredentialsApiConfigMixin
actual = get_user_credentials(self.user)
self.assertEqual(actual, self.CREDENTIALS_API_RESPONSE['results'])
@httpretty.activate
def test_get_user_credentials_caching(self):
"""Verify that when enabled, the cache is used for non-staff users."""
self.create_credentials_config(cache_ttl=1)
self.mock_credentials_api(self.user)
# Warm up the cache.
get_user_credentials(self.user)
# Hit the cache.
get_user_credentials(self.user)
# Verify only one request was made.
self.assertEqual(len(httpretty.httpretty.latest_requests), 1)
staff_user = UserFactory(is_staff=True)
# Hit the Credentials API twice.
for _ in range(2):
get_user_credentials(staff_user)
# Verify that three requests have been made (one for student, two for staff).
self.assertEqual(len(httpretty.httpretty.latest_requests), 3)
def test_get_user_program_credentials_issuance_disable(self):
"""Verify that user program credentials cannot be retrieved if issuance is disabled."""
self.create_credentials_config(enable_learner_issuance=False)
......@@ -50,6 +77,28 @@ class TestCredentialsRetrieval(ProgramsApiConfigMixin, CredentialsApiConfigMixin
self.assertEqual(actual, [])
@httpretty.activate
def test_get_user_programs_credentials(self):
"""Verify program credentials data can be retrieved and parsed correctly."""
# create credentials and program configuration
credentials_config = self.create_credentials_config()
self.create_programs_config()
# Mocking the API responses from programs and credentials
self.mock_programs_api()
self.mock_credentials_api(self.user, reset_url=False)
actual = get_user_program_credentials(self.user)
expected = self.PROGRAMS_API_RESPONSE['results']
expected[0]['credential_url'] = \
credentials_config.public_service_url + 'credentials/' + self.PROGRAMS_CREDENTIALS_DATA[0]['uuid']
expected[1]['credential_url'] = \
credentials_config.public_service_url + 'credentials/' + self.PROGRAMS_CREDENTIALS_DATA[1]['uuid']
# checking response from API is as expected
self.assertEqual(len(actual), 2)
self.assertEqual(actual, expected)
@httpretty.activate
def test_get_user_program_credentials_revoked(self):
"""Verify behavior if credential revoked."""
self.create_credentials_config()
......@@ -68,21 +117,3 @@ class TestCredentialsRetrieval(ProgramsApiConfigMixin, CredentialsApiConfigMixin
self.mock_credentials_api(self.user, data=credential_data)
actual = get_user_program_credentials(self.user)
self.assertEqual(actual, [])
@httpretty.activate
def test_get_user_programs_credentials(self):
"""Verify program credentials data can be retrieved and parsed correctly."""
credentials_config = self.create_credentials_config()
self.create_programs_config()
self.mock_programs_api()
self.mock_credentials_api(self.user, reset_url=False)
actual = get_user_program_credentials(self.user)
expected = self.PROGRAMS_API_RESPONSE['results']
expected[0]['credential_url'] = \
credentials_config.public_service_url + 'credentials/' + self.PROGRAMS_CREDENTIALS_DATA[0]['uuid']
expected[1]['credential_url'] = \
credentials_config.public_service_url + 'credentials/' + self.PROGRAMS_CREDENTIALS_DATA[1]['uuid']
self.assertEqual(len(actual), 2)
self.assertEqual(actual, expected)
httpretty.reset()
openedx/core/djangoapps/credentials/utils.py
......@@ -4,7 +4,7 @@ import logging
from openedx.core.djangoapps.credentials.models import CredentialsApiConfig
from openedx.core.djangoapps.programs.utils import get_programs_for_credentials
from openedx.core.lib.api_utils import get_api_data
from openedx.core.lib.edx_api_utils import get_edx_api_data
log = logging.getLogger(__name__)
......@@ -20,8 +20,13 @@ def get_user_credentials(user):
"""
credential_configuration = CredentialsApiConfig.current()
user_query = {'username': user.username}
credentials = get_api_data(
credential_configuration, user, credential_configuration.API_NAME, 'user_credentials', querystring=user_query
# Bypass caching for staff users, who may be generating credentials and
# want to see them displayed immediately.
use_cache = credential_configuration.is_cache_enabled and not user.is_staff
cache_key = credential_configuration.CACHE_KEY + '.' + user.username if use_cache else None
credentials = get_edx_api_data(
credential_configuration, user, 'user_credentials', querystring=user_query, cache_key=cache_key
)
return credentials
......
openedx/core/djangoapps/organization_api/__init__.py 0 → 100644
"""
Organization API.
WARNING: This API is intended to move into the 'edx-organizations' repo.
https://github.com/edx/edx-organizations
"""
openedx/core/djangoapps/organization_api/api/views.py 0 → 100644
"""
Organizations API views.
"""
from rest_framework import permissions
from rest_framework.authentication import SessionAuthentication
from rest_framework.response import Response
from rest_framework.status import HTTP_404_NOT_FOUND
from rest_framework.views import APIView
from rest_framework_oauth.authentication import OAuth2Authentication
from util.organizations_helpers import get_organization_by_short_name
class OrganizationsView(APIView):
"""
View to get organization information.
"""
authentication_classes = (OAuth2Authentication, SessionAuthentication,)
permission_classes = (permissions.IsAuthenticated,)
def get(self, request, organization_key):
"""
Return organization information related to provided organization
key/short_name.
"""
organization = get_organization_by_short_name(organization_key)
if organization:
logo = organization.get('logo')
organization_data = {
'name': organization.get('name', ''),
'short_name': organization.get('short_name', ''),
'description': organization.get('description', ''),
'logo': request.build_absolute_uri(logo.url) if logo else ''
}
return Response(organization_data)
return Response(status=HTTP_404_NOT_FOUND)
openedx/core/djangoapps/organization_api/tests/test_views.py 0 → 100644
"""
Tests for organization API.
"""
import ddt
import json
import unittest
from django.conf import settings
from django.core.urlresolvers import reverse, NoReverseMatch
from django.test import TestCase
from oauth2_provider.tests.factories import AccessTokenFactory, ClientFactory
from student.tests.factories import UserFactory
from util import organizations_helpers
@ddt.ddt
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
class OrganizationsAPITests(TestCase):
"""
Tests for the organizations API endpoints.
GET /api/organizations/v1/organization/:org_key/
"""
def setUp(self):
"""
Test setup for the organizations API.
"""
super(OrganizationsAPITests, self).setUp()
self.user_password = 'password'
self.user = UserFactory(password=self.user_password)
self.test_org_key = 'test_organization'
self.test_org_url = self._generate_org_url(self.test_org_key)
def _create_test_organization(self, org_key=None):
"""
Helper method to create a test organization with the provide 'org_key' and
returns the url to access it.
"""
if org_key is None:
org_key = self.test_org_key
test_organization_data = {
'name': 'Test Organization',
'short_name': org_key,
'description': 'Test Organization Description',
'logo': '/test_logo.png/'
}
organizations_helpers.add_organization(organization_data=test_organization_data)
return self._generate_org_url(org_key)
def _generate_org_url(self, org_key):
"""
Helper method to generate the url to get organization data for a
specific organization key.
"""
return reverse(
'organization_api:get_organization', kwargs={'organization_key': org_key}
)
def test_authentication_required(self):
"""
Verify that the endpoint requires authentication.
"""
response = self.client.get(self.test_org_url)
self.assertEqual(response.status_code, 401)
def test_session_auth(self):
"""
Verify that the endpoint supports session authentication.
"""
self.client.login(username=self.user.username, password=self.user_password)
response = self.client.get(self.test_org_url)
# verify that the test org does not exist
self.assertEqual(response.status_code, 404)
# add a test organization
self._create_test_organization()
# verify that the organization api return data in correct format
response = self.client.get(self.test_org_url)
self.assertEqual(response.status_code, 200)
expected_output = {
'name': 'Test Organization',
'short_name': 'test_organization',
'description': 'Test Organization Description',
'logo': 'http://testserver/test_logo.png/'
}
self.assertEqual(json.loads(response.content), expected_output)
def test_oauth(self):
"""
Verify that the organization API supports OAuth.
"""
oauth_client = ClientFactory.create()
access_token = AccessTokenFactory.create(user=self.user, client=oauth_client).token
headers = {
'HTTP_AUTHORIZATION': 'Bearer ' + access_token
}
response = self.client.get(self.test_org_url, **headers)
# verify that the test org does not exist
self.assertEqual(response.status_code, 404)
# add a test organization
self._create_test_organization()
# verify that the organization api return data in correct format
response = self.client.get(self.test_org_url, **headers)
self.assertEqual(response.status_code, 200)
expected_output = {
'name': 'Test Organization',
'short_name': 'test_organization',
'description': 'Test Organization Description',
'logo': 'http://testserver/test_logo.png/'
}
self.assertEqual(json.loads(response.content), expected_output)
@ddt.data("test_org's", "test_org*", "test(org)", "!test", "test org")
def test_with_invalid_org_key(self, invalid_org_key):
"""
Verify that organization url does not match for invalid org key.
"""
with self.assertRaises(NoReverseMatch):
self._generate_org_url(invalid_org_key)
openedx/core/djangoapps/organization_api/urls.py 0 → 100644
"""
URLs for the organization app.
"""
from django.conf.urls import patterns, url
from openedx.core.djangoapps.organization_api.api import views
ORGANIZATION_KEY_PATTERN = r"(?P<organization_key>((?![\^'\!\(\)\*\s]).)*)"
urlpatterns = patterns(
'',
url(
r'^v0/organization/{}/$'.format(ORGANIZATION_KEY_PATTERN),
views.OrganizationsView.as_view(),
name='get_organization'
),
)
openedx/core/djangoapps/programs/tests/test_utils.py
......@@ -2,6 +2,7 @@
from django.core.cache import cache
from django.test import TestCase
import httpretty
import mock
from oauth2_provider.tests.factories import ClientFactory
from provider.constants import CONFIDENTIAL
......@@ -41,6 +42,56 @@ class TestProgramRetrieval(ProgramsApiConfigMixin, ProgramsDataMixin,
self.assertEqual(len(httpretty.httpretty.latest_requests), 1)
@httpretty.activate
def test_get_programs_caching(self):
"""Verify that when enabled, the cache is used for non-staff users."""
self.create_programs_config(cache_ttl=1)
self.mock_programs_api()
# Warm up the cache.
get_programs(self.user)
# Hit the cache.
get_programs(self.user)
# Verify only one request was made.
self.assertEqual(len(httpretty.httpretty.latest_requests), 1)
staff_user = UserFactory(is_staff=True)
# Hit the Programs API twice.
for _ in range(2):
get_programs(staff_user)
# Verify that three requests have been made (one for student, two for staff).
self.assertEqual(len(httpretty.httpretty.latest_requests), 3)
def test_get_programs_programs_disabled(self):
"""Verify behavior when programs is disabled."""
self.create_programs_config(enabled=False)
actual = get_programs(self.user)
self.assertEqual(actual, [])
@mock.patch('edx_rest_api_client.client.EdxRestApiClient.__init__')
def test_get_programs_client_initialization_failure(self, mock_init):
"""Verify behavior when API client fails to initialize."""
self.create_programs_config()
mock_init.side_effect = Exception
actual = get_programs(self.user)
self.assertEqual(actual, [])
self.assertTrue(mock_init.called)
@httpretty.activate
def test_get_programs_data_retrieval_failure(self):
"""Verify behavior when data can't be retrieved from Programs."""
self.create_programs_config()
self.mock_programs_api(status_code=500)
actual = get_programs(self.user)
self.assertEqual(actual, [])
@httpretty.activate
def test_get_programs_for_dashboard(self):
"""Verify programs data can be retrieved and parsed correctly."""
self.create_programs_config()
......
openedx/core/djangoapps/programs/utils.py
......@@ -4,7 +4,7 @@ from urlparse import urljoin
from openedx.core.djangoapps.credentials.models import CredentialsApiConfig
from openedx.core.djangoapps.programs.models import ProgramsApiConfig
from openedx.core.lib.api_utils import get_api_data
from openedx.core.lib.edx_api_utils import get_edx_api_data
log = logging.getLogger(__name__)
......@@ -24,9 +24,11 @@ def get_programs(user):
"""
programs_config = ProgramsApiConfig.current()
# Bypass caching for staff users, who may be creating Programs and want to see them displayed immediately.
use_cache = programs_config.is_cache_enabled and not user.is_staff
return get_api_data(programs_config, user, programs_config.API_NAME, 'programs', use_cache=use_cache)
# Bypass caching for staff users, who may be creating Programs and want
# to see them displayed immediately.
cache_key = programs_config.CACHE_KEY if programs_config.is_cache_enabled and not user.is_staff else None
return get_edx_api_data(programs_config, user, 'programs', cache_key=cache_key)
def get_programs_for_dashboard(user, course_keys):
......
openedx/core/lib/api_utils.py → openedx/core/lib/edx_api_utils.py
......@@ -11,19 +11,20 @@ from openedx.core.lib.token_utils import get_id_token
log = logging.getLogger(__name__)
def get_api_data(api_config, user, api_name, resource, querystring=None, use_cache=False):
"""Fetch the data from the API using provided API Configuration and
resource.
def get_edx_api_data(api_config, user, resource, querystring=None, cache_key=None):
"""Fetch data from an API using provided API configuration and resource
name.
Arguments:
api_config: The configuration which will be user for requesting data.
api_config (ConfigurationModel): The configuration model governing
interaction with the API.
user (User): The user to authenticate as when requesting data.
api_name: Name fo the api to be use for logging.
resource: API resource to from where data will be requested.
querystring: Querystring parameters that might be required to request
data.
use_cache: Will be used to decide whether to cache the response data
or not.
resource(str): Name of the API resource for which data is being
requested.
querystring(dict): Querystring parameters that might be required to
request data.
cache_key(str): Where to cache retrieved data. Omitting this will cause the
cache to be bypassed.
Returns:
list of dict, representing data returned by the API.
......@@ -31,23 +32,19 @@ def get_api_data(api_config, user, api_name, resource, querystring=None, use_cac
no_data = []
if not api_config.enabled:
log.warning('%s configuration is disabled.', api_name)
log.warning('%s configuration is disabled.', api_config.API_NAME)
return no_data
if use_cache:
if api_config.CACHE_KEY:
cached = cache.get(api_config.CACHE_KEY)
if cached is not None:
return cached
else:
log.warning('No cache key available for %s configuration.', api_name)
return no_data
if cache_key:
cached = cache.get(cache_key)
if cached is not None:
return cached
try:
jwt = get_id_token(user, api_config.OAUTH2_CLIENT_NAME)
api = EdxRestApiClient(api_config.internal_api_url, jwt=jwt)
except Exception: # pylint: disable=broad-except
log.exception('Failed to initialize the %s API client.', api_name)
log.exception('Failed to initialize the %s API client.', api_config.API_NAME)
return no_data
try:
......@@ -63,10 +60,10 @@ def get_api_data(api_config, user, api_name, resource, querystring=None, use_cac
results += response.get('results', no_data)
next_page = response.get('next', None)
except Exception: # pylint: disable=broad-except
log.exception('Failed to retrieve data from the %s API.', api_name)
log.exception('Failed to retrieve data from the %s API.', api_config.API_NAME)
return no_data
if use_cache:
cache.set(api_config.CACHE_KEY, results, api_config.cache_ttl)
if cache_key:
cache.set(cache_key, results, api_config.cache_ttl)
return results
openedx/core/lib/tests/test_api_utils.py → openedx/core/lib/tests/test_edx_api_utils.py
......@@ -10,13 +10,13 @@ from openedx.core.djangoapps.credentials.models import CredentialsApiConfig
from openedx.core.djangoapps.credentials.tests.mixins import CredentialsApiConfigMixin, CredentialsDataMixin
from openedx.core.djangoapps.programs.models import ProgramsApiConfig
from openedx.core.djangoapps.programs.tests.mixins import ProgramsApiConfigMixin, ProgramsDataMixin
from openedx.core.lib.api_utils import get_api_data
from openedx.core.lib.edx_api_utils import get_edx_api_data
from student.tests.factories import UserFactory
class TestApiDataRetrieval(CredentialsApiConfigMixin, CredentialsDataMixin, ProgramsApiConfigMixin, ProgramsDataMixin,
TestCase):
"""Test data retrieval from the api util function."""
"""Test utility for API data retrieval."""
def setUp(self):
super(TestApiDataRetrieval, self).setUp()
ClientFactory(name=CredentialsApiConfig.OAUTH2_CLIENT_NAME, client_type=CONFIDENTIAL)
......@@ -26,12 +26,12 @@ class TestApiDataRetrieval(CredentialsApiConfigMixin, CredentialsDataMixin, Prog
cache.clear()
@httpretty.activate
def test_get_api_data_programs(self):
"""Verify programs data can be retrieve using get_api_data."""
def test_get_edx_api_data_programs(self):
"""Verify programs data can be retrieved using get_edx_api_data."""
program_config = self.create_programs_config()
self.mock_programs_api()
actual = get_api_data(program_config, self.user, 'programs', 'programs')
actual = get_edx_api_data(program_config, self.user, 'programs')
self.assertEqual(
actual,
self.PROGRAMS_API_RESPONSE['results']
......@@ -40,75 +40,54 @@ class TestApiDataRetrieval(CredentialsApiConfigMixin, CredentialsDataMixin, Prog
# Verify the API was actually hit (not the cache).
self.assertEqual(len(httpretty.httpretty.latest_requests), 1)
@httpretty.activate
def test_get_api_data_credentials(self):
"""Verify credentials data can be retrieve using get_api_data."""
credentials_config = self.create_credentials_config()
self.mock_credentials_api(self.user)
querystring = {'username': self.user.username}
actual = get_api_data(credentials_config, self.user, 'credentials', 'user_credentials', querystring=querystring)
self.assertEqual(
actual,
self.CREDENTIALS_API_RESPONSE['results']
)
def test_get_api_data_disable_config(self):
"""Verify no data is retrieve if configuration is disabled."""
def test_get_edx_api_data_disable_config(self):
"""Verify no data is retrieved if configuration is disabled."""
program_config = self.create_programs_config(enabled=False)
actual = get_api_data(program_config, self.user, 'programs', 'programs')
actual = get_edx_api_data(program_config, self.user, 'programs')
self.assertEqual(actual, [])
@httpretty.activate
def test_get_api_data_cache(self):
def test_get_edx_api_data_cache(self):
"""Verify that when enabled, the cache is used."""
program_config = self.create_programs_config(cache_ttl=1)
self.mock_programs_api()
# Warm up the cache.
get_api_data(program_config, self.user, 'programs', 'programs', use_cache=True)
get_edx_api_data(program_config, self.user, 'programs', cache_key='test.key')
# Hit the cache.
get_api_data(program_config, self.user, 'programs', 'programs', use_cache=True)
get_edx_api_data(program_config, self.user, 'programs', cache_key='test.key')
# Verify only one request was made.
self.assertEqual(len(httpretty.httpretty.latest_requests), 1)
def test_get_api_data_without_cache_key(self):
"""Verify that when cache enabled without cache key then no data is retrieved."""
ProgramsApiConfig.CACHE_KEY = None
program_config = self.create_programs_config(cache_ttl=1)
actual = get_api_data(program_config, self.user, 'programs', 'programs', use_cache=True)
self.assertEqual(actual, [])
@mock.patch('edx_rest_api_client.client.EdxRestApiClient.__init__')
def test_get_api_data_client_initialization_failure(self, mock_init):
def test_get_edx_api_data_client_initialization_failure(self, mock_init):
"""Verify behavior when API client fails to initialize."""
program_config = self.create_programs_config()
mock_init.side_effect = Exception
actual = get_api_data(program_config, self.user, 'programs', 'programs')
actual = get_edx_api_data(program_config, self.user, 'programs')
self.assertEqual(actual, [])
self.assertTrue(mock_init.called)
@httpretty.activate
def test_get_api_data_retrieval_failure(self):
def test_get_edx_api_data_retrieval_failure(self):
"""Verify behavior when data can't be retrieved from API."""
program_config = self.create_programs_config()
self.mock_programs_api(status_code=500)
actual = get_api_data(program_config, self.user, 'programs', 'programs')
actual = get_edx_api_data(program_config, self.user, 'programs')
self.assertEqual(actual, [])
@httpretty.activate
def test_get_api_data_multiple_page(self):
def test_get_edx_api_data_multiple_page(self):
"""Verify that all data is retrieve for multiple page response."""
credentials_config = self.create_credentials_config()
self.mock_credentials_api(self.user, is_next_page=True)
querystring = {'username': self.user.username}
actual = get_api_data(credentials_config, self.user, 'credentials', 'user_credentials', querystring=querystring)
actual = get_edx_api_data(credentials_config, self.user, 'user_credentials', querystring=querystring)
expected_data = self.CREDENTIALS_NEXT_API_RESPONSE['results'] + self.CREDENTIALS_API_RESPONSE['results']
self.assertEqual(actual, expected_data)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment