Skip to content

E
edx-platform
Commit 9be31918 by Renzo Lucioni Committed by GitHub
Options

Merge pull request #14462 from edx/renzo/programs-from-catalog 

Load all programs from the catalog
parents 54e4bcfa e7771148
Showing with 958 additions and 1722 deletions
common/djangoapps/student/tests/tests.py
......@@ -14,7 +14,6 @@ from django.contrib.auth.models import User, AnonymousUser
from django.core.urlresolvers import reverse
from django.test import TestCase, override_settings
from django.test.client import Client
from edx_oauth2_provider.tests.factories import ClientFactory
import httpretty
from markupsafe import escape
from mock import Mock, patch
......@@ -30,11 +29,15 @@ from certificates.tests.factories import GeneratedCertificateFactory # pylint:
from config_models.models import cache
from course_modes.models import CourseMode
from lms.djangoapps.verify_student.models import SoftwareSecurePhotoVerification
from openedx.core.djangolib.testing.utils import CacheIsolationTestCase
from openedx.core.djangoapps.programs.models import ProgramsApiConfig
from openedx.core.djangoapps.programs.tests import factories as programs_factories
from openedx.core.djangoapps.catalog.tests.factories import (
generate_course_run_key,
ProgramFactory,
CourseFactory as CatalogCourseFactory,
CourseRunFactory,
)
from openedx.core.djangoapps.programs.tests.mixins import ProgramsApiConfigMixin
from openedx.core.djangoapps.site_configuration.tests.mixins import SiteMixin
from openedx.core.djangolib.testing.utils import CacheIsolationTestCase, skip_unless_lms
import shoppingcart # pylint: disable=import-error
from student.models import (
anonymous_id_for_user, user_by_anonymous_id, CourseEnrollment,
......@@ -988,11 +991,10 @@ class AnonymousLookupTable(ModuleStoreTestCase):
@attr(shard=3)
@httpretty.activate
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
@skip_unless_lms
@patch('openedx.core.djangoapps.programs.utils.get_programs')
class RelatedProgramsTests(ProgramsApiConfigMixin, SharedModuleStoreTestCase):
"""Tests verifying that related programs appear on the course dashboard."""
url = None
maxDiff = None
password = 'test'
related_programs_preface = 'Related Programs'
......@@ -1004,18 +1006,6 @@ class RelatedProgramsTests(ProgramsApiConfigMixin, SharedModuleStoreTestCase):
cls.user = UserFactory()
cls.course = CourseFactory()
cls.enrollment = CourseEnrollmentFactory(user=cls.user, course_id=cls.course.id) # pylint: disable=no-member
ClientFactory(name=ProgramsApiConfig.OAUTH2_CLIENT_NAME, client_type=CONFIDENTIAL)
cls.organization = programs_factories.Organization()
run_mode = programs_factories.RunMode(course_key=unicode(cls.course.id)) # pylint: disable=no-member
course_code = programs_factories.CourseCode(run_modes=[run_mode])
cls.programs = [
programs_factories.Program(
organizations=[cls.organization],
course_codes=[course_code]
) for __ in range(2)
]
def setUp(self):
super(RelatedProgramsTests, self).setUp()
......@@ -1025,14 +1015,9 @@ class RelatedProgramsTests(ProgramsApiConfigMixin, SharedModuleStoreTestCase):
self.create_programs_config()
self.client.login(username=self.user.username, password=self.password)
def mock_programs_api(self, data):
"""Helper for mocking out Programs API URLs."""
self.assertTrue(httpretty.is_enabled(), msg='httpretty must be enabled to mock Programs API calls.')
url = ProgramsApiConfig.current().internal_api_url.strip('/') + '/programs/'
body = json.dumps({'results': data})
httpretty.register_uri(httpretty.GET, url, body=body, content_type='application/json')
course_run = CourseRunFactory(key=unicode(self.course.id)) # pylint: disable=no-member
course = CatalogCourseFactory(course_runs=[course_run])
self.programs = [ProgramFactory(courses=[course]) for __ in range(2)]
def assert_related_programs(self, response, are_programs_present=True):
"""Assertion for verifying response contents."""
......@@ -1045,42 +1030,40 @@ class RelatedProgramsTests(ProgramsApiConfigMixin, SharedModuleStoreTestCase):
def expected_link_text(self, program):
"""Construct expected dashboard link text."""
return u'{name} {category}'.format(name=program['name'], category=program['category'])
return u'{title} {type}'.format(title=program['title'], type=program['type'])
def test_related_programs_listed(self):
"""Verify that related programs are listed when the programs API returns data."""
self.mock_programs_api(self.programs)
def test_related_programs_listed(self, mock_get_programs):
"""Verify that related programs are listed when available."""
mock_get_programs.return_value = self.programs
response = self.client.get(self.url)
self.assert_related_programs(response)
def test_no_data_no_programs(self):
"""Verify that related programs aren't listed if the programs API returns no data."""
self.mock_programs_api([])
def test_no_data_no_programs(self, mock_get_programs):
"""Verify that related programs aren't listed when none are available."""
mock_get_programs.return_value = []
response = self.client.get(self.url)
self.assert_related_programs(response, are_programs_present=False)
def test_unrelated_program_not_listed(self):
def test_unrelated_program_not_listed(self, mock_get_programs):
"""Verify that unrelated programs don't appear in the listing."""
run_mode = programs_factories.RunMode(course_key='some/nonexistent/run')
course_code = programs_factories.CourseCode(run_modes=[run_mode])
nonexistent_course_run_id = generate_course_run_key()
unrelated_program = programs_factories.Program(
organizations=[self.organization],
course_codes=[course_code]
)
course_run = CourseRunFactory(key=nonexistent_course_run_id)
course = CatalogCourseFactory(course_runs=[course_run])
unrelated_program = ProgramFactory(courses=[course])
self.mock_programs_api(self.programs + [unrelated_program])
mock_get_programs.return_value = self.programs + [unrelated_program]
response = self.client.get(self.url)
self.assert_related_programs(response)
self.assertNotContains(response, unrelated_program['name'])
self.assertNotContains(response, unrelated_program['title'])
def test_program_title_unicode(self):
def test_program_title_unicode(self, mock_get_programs):
"""Verify that the dashboard can deal with programs whose titles contain Unicode."""
self.programs[0]['name'] = u'Bases matemáticas para estudiar ingeniería'
self.mock_programs_api(self.programs)
self.programs[0]['title'] = u'Bases matemáticas para estudiar ingeniería'
mock_get_programs.return_value = self.programs
response = self.client.get(self.url)
self.assert_related_programs(response)
......
common/djangoapps/student/views.py
......@@ -13,7 +13,6 @@ from django.views.generic import TemplateView
from pytz import UTC
from requests import HTTPError
from ipware.ip import get_ip
import waffle
import edx_oauth2_provider
from django.conf import settings
......@@ -125,12 +124,13 @@ from notification_prefs.views import enable_notifications
from openedx.core.djangoapps.credit.email_utils import get_credit_provider_display_names, make_providers_strings
from openedx.core.djangoapps.lang_pref import LANGUAGE_KEY
from openedx.core.djangoapps.programs import utils as programs_utils
from openedx.core.djangoapps.catalog.utils import munge_catalog_program
from openedx.core.djangoapps.programs.models import ProgramsApiConfig
from openedx.core.djangoapps.programs.utils import ProgramProgressMeter
from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
from openedx.core.djangoapps.theming import helpers as theming_helpers
from openedx.core.djangoapps.user_api.preferences import api as preferences_api
from openedx.core.djangoapps.catalog.utils import get_programs_data
from openedx.core.djangoapps.catalog.utils import get_programs_with_type_logo
log = logging.getLogger("edx.student")
......@@ -217,7 +217,7 @@ def index(request, extra_context=None, user=AnonymousUser()):
# for edx-pattern-library is added.
if configuration_helpers.get_value("DISPLAY_PROGRAMS_ON_MARKETING_PAGES",
settings.FEATURES.get("DISPLAY_PROGRAMS_ON_MARKETING_PAGES")):
programs_list = get_programs_data(user)
programs_list = get_programs_with_type_logo()
context["programs_list"] = programs_list
......@@ -664,12 +664,14 @@ def dashboard(request):
and has_access(request.user, 'view_courseware_with_prerequisites', enrollment.course_overview)
)
# Find programs associated with courses being displayed. This information
# Find programs associated with course runs being displayed. This information
# is passed in the template context to allow rendering of program-related
# information on the dashboard.
use_catalog = waffle.switch_is_active('get_programs_from_catalog')
meter = programs_utils.ProgramProgressMeter(user, enrollments=course_enrollments, use_catalog=use_catalog)
programs_by_run = meter.engaged_programs(by_run=True)
meter = ProgramProgressMeter(user, enrollments=course_enrollments)
inverted_programs = meter.invert_programs()
for program_list in inverted_programs.itervalues():
program_list[:] = [munge_catalog_program(program) for program in program_list]
# Construct a dictionary of course mode information
# used to render the course list. We re-use the course modes dict
......@@ -793,7 +795,7 @@ def dashboard(request):
'order_history_list': order_history_list,
'courses_requirements_not_met': courses_requirements_not_met,
'nav_hidden': True,
'programs_by_run': programs_by_run,
'programs_by_run': inverted_programs,
'show_program_listing': ProgramsApiConfig.current().show_program_listing,
'disable_courseware_js': True,
'display_course_modes_on_dashboard': enable_verified_certificates and display_course_modes_on_dashboard,
......
common/djangoapps/terrain/stubs/catalog.py
"""
Stub implementation of catalog service for acceptance tests
"""
# pylint: disable=invalid-name, missing-docstring
import re
import urlparse
from .http import StubHttpRequestHandler, StubHttpService
class StubCatalogServiceHandler(StubHttpRequestHandler): # pylint: disable=missing-docstring
class StubCatalogServiceHandler(StubHttpRequestHandler):
def do_GET(self): # pylint: disable=invalid-name, missing-docstring
def do_GET(self):
pattern_handlers = {
r'/api/v1/programs/$': self.get_programs,
r'/api/v1/course_runs/(?P<course_id>[^/+]+(/|\+)[^/+]+(/|\+)[^/?]+)/$': self.get_course_run,
r'/api/v1/programs/$': self.program_list,
r'/api/v1/programs/([0-9a-f-]+)/$': self.program_detail,
}
if self.match_pattern(pattern_handlers):
return
self.send_response(404, content="404 Not Found")
self.send_response(404, content='404 Not Found')
def match_pattern(self, pattern_handlers):
"""
Find the correct handler method given the path info from the HTTP request.
"""
path = urlparse.urlparse(self.path).path
for pattern in pattern_handlers:
for pattern, handler in pattern_handlers.items():
match = re.match(pattern, path)
if match:
pattern_handlers[pattern](*match.groups())
handler(*match.groups())
return True
return None
def get_programs(self):
"""
Stubs the catalog's programs endpoint.
"""
def program_list(self):
"""Stub the catalog's program list endpoint."""
programs = self.server.config.get('catalog.programs', [])
self.send_json_response(programs)
def get_course_run(self, course_id):
"""
Stubs the catalog's course run endpoint.
"""
course_run = self.server.config.get('course_run.{}'.format(course_id), [])
self.send_json_response(course_run)
def program_detail(self, program_uuid):
"""Stub the catalog's program detail endpoint."""
program = self.server.config.get('catalog.programs.' + program_uuid)
self.send_json_response(program)
class StubCatalogService(StubHttpService): # pylint: disable=missing-docstring
class StubCatalogService(StubHttpService):
HANDLER_CLASS = StubCatalogServiceHandler
common/djangoapps/terrain/stubs/programs.py deleted 100644 → 0
"""
Stub implementation of programs service for acceptance tests
"""
import re
import urlparse
from .http import StubHttpRequestHandler, StubHttpService
class StubProgramsServiceHandler(StubHttpRequestHandler): # pylint: disable=missing-docstring
def do_GET(self): # pylint: disable=invalid-name, missing-docstring
pattern_handlers = {
r'/api/v1/programs/$': self.get_programs_list,
r'/api/v1/programs/(\d+)/$': self.get_program_details,
}
if self.match_pattern(pattern_handlers):
return
self.send_response(404, content="404 Not Found")
def match_pattern(self, pattern_handlers):
"""
Find the correct handler method given the path info from the HTTP request.
"""
path = urlparse.urlparse(self.path).path
for pattern in pattern_handlers:
match = re.match(pattern, path)
if match:
pattern_handlers[pattern](*match.groups())
return True
return None
def get_programs_list(self):
"""
Stubs the programs list endpoint.
"""
programs = self.server.config.get('programs', [])
self.send_json_response(programs)
def get_program_details(self, program_id):
"""
Stubs a program details endpoint.
"""
program = self.server.config.get('programs.{}'.format(program_id), [])
self.send_json_response(program)
class StubProgramsService(StubHttpService): # pylint: disable=missing-docstring
HANDLER_CLASS = StubProgramsServiceHandler
common/djangoapps/terrain/stubs/start.py
......@@ -12,7 +12,6 @@ from .youtube import StubYouTubeService
from .lti import StubLtiService
from .video_source import VideoSourceHttpService
from .edxnotes import StubEdxNotesService
from .programs import StubProgramsService
from .catalog import StubCatalogService
......@@ -25,7 +24,6 @@ SERVICES = {
'lti': StubLtiService,
'video': VideoSourceHttpService,
'edxnotes': StubEdxNotesService,
'programs': StubProgramsService,
'ecommerce': StubEcommerceService,
'catalog': StubCatalogService,
}
......
common/test/acceptance/fixtures/__init__.py
......@@ -19,8 +19,5 @@ COMMENTS_STUB_URL = os.environ.get('comments_url', 'http://localhost:4567')
# Get the URL of the EdxNotes service stub used in the test
EDXNOTES_STUB_URL = os.environ.get('edxnotes_url', 'http://localhost:8042')
# Get the URL of the Programs service stub used in the test
PROGRAMS_STUB_URL = os.environ.get('programs_url', 'http://localhost:8090')
# Get the URL of the Catalog service stub used in the test
CATALOG_STUB_URL = os.environ.get('catalog_url', 'http://localhost:8091')
common/test/acceptance/fixtures/catalog.py
......@@ -13,31 +13,31 @@ class CatalogFixture(object):
"""
Interface to set up mock responses from the Catalog stub server.
"""
def install_programs(self, programs):
def install_programs(self, data):
"""Set response data for the catalog's course run API."""
key = 'catalog.programs'
if isinstance(data, dict):
key += '.' + data['uuid']
requests.put(
'{}/set_config'.format(CATALOG_STUB_URL),
data={key: json.dumps(programs)},
data={key: json.dumps(data)},
)
def install_course_run(self, course_run):
"""Set response data for the catalog's course run API."""
key = 'catalog.{}'.format(course_run['key'])
else:
requests.put(
'{}/set_config'.format(CATALOG_STUB_URL),
data={key: json.dumps(course_run)},
data={key: json.dumps({'results': data})},
)
class CatalogConfigMixin(object):
class CatalogIntegrationMixin(object):
"""Mixin providing a method used to configure the catalog integration."""
def set_catalog_configuration(self, is_enabled=False, service_url=CATALOG_STUB_URL):
"""Dynamically adjusts the catalog config model during tests."""
def set_catalog_integration(self, is_enabled=False, service_username=None):
"""Use this to change the catalog integration config model during tests."""
ConfigModelFixture('/config/catalog', {
'enabled': is_enabled,
'internal_api_url': '{}/api/v1/'.format(service_url),
'internal_api_url': '{}/api/v1/'.format(CATALOG_STUB_URL),
'cache_ttl': 0,
'service_username': service_username,
}).install()
common/test/acceptance/fixtures/programs.py
"""
Tools to create programs-related data for use in bok choy tests.
"""
import json
import requests
from common.test.acceptance.fixtures import PROGRAMS_STUB_URL
from common.test.acceptance.fixtures.config import ConfigModelFixture
class ProgramsFixture(object):
"""
Interface to set up mock responses from the Programs stub server.
"""
def install_programs(self, programs, is_list=True):
"""Sets the response data for Programs API endpoints."""
if is_list:
key = 'programs'
api_result = {'results': programs}
else:
program = programs[0]
key = 'programs.{}'.format(program['id'])
api_result = program
requests.put(
'{}/set_config'.format(PROGRAMS_STUB_URL),
data={key: json.dumps(api_result)},
)
class ProgramsConfigMixin(object):
"""Mixin providing a method used to configure the programs feature."""
def set_programs_api_configuration(self, is_enabled=False, api_version=1, api_url=PROGRAMS_STUB_URL):
def set_programs_api_configuration(self, is_enabled=False, api_version=1):
"""Dynamically adjusts the Programs config model during tests."""
ConfigModelFixture('/config/programs', {
'enabled': is_enabled,
'api_version_number': api_version,
'internal_service_url': api_url,
'public_service_url': api_url,
'cache_ttl': 0,
'marketing_path': '/foo',
'enable_student_dashboard': is_enabled,
'enable_studio_tab': is_enabled,
'enable_certification': is_enabled,
'program_listing_enabled': is_enabled,
'program_details_enabled': is_enabled,
......
common/test/acceptance/pages/lms/programs.py
"""LMS-hosted Programs pages"""
from uuid import uuid4
from bok_choy.page_object import PageObject
from common.test.acceptance.pages.lms import BASE_URL
......@@ -24,8 +26,8 @@ class ProgramListingPage(PageObject):
class ProgramDetailsPage(PageObject):
"""Program details page."""
program_id = 123
url = BASE_URL + '/dashboard/programs/{}/program-name/'.format(program_id)
program_uuid = str(uuid4())
url = '{base}/dashboard/programs/{program_uuid}/'.format(base=BASE_URL, program_uuid=program_uuid)
def is_browser_on_page(self):
return self.q(css='.js-program-details-wrapper').present
common/test/acceptance/tests/lms/test_programs.py
"""Acceptance tests for LMS-hosted Programs pages"""
from nose.plugins.attrib import attr
from common.test.acceptance.fixtures.catalog import CatalogFixture, CatalogConfigMixin
from common.test.acceptance.fixtures.programs import ProgramsFixture, ProgramsConfigMixin
from common.test.acceptance.fixtures.catalog import CatalogFixture, CatalogIntegrationMixin
from common.test.acceptance.fixtures.programs import ProgramsConfigMixin
from common.test.acceptance.fixtures.course import CourseFixture
from common.test.acceptance.tests.helpers import UniqueCourseTest
from common.test.acceptance.pages.lms.auto_auth import AutoAuthPage
from common.test.acceptance.pages.lms.programs import ProgramListingPage, ProgramDetailsPage
from openedx.core.djangoapps.catalog.tests import factories as catalog_factories
from openedx.core.djangoapps.programs.tests import factories as program_factories
from openedx.core.djangoapps.catalog.tests.factories import (
ProgramFactory, CourseFactory, CourseRunFactory
)
class ProgramPageBase(ProgramsConfigMixin, CatalogConfigMixin, UniqueCourseTest):
class ProgramPageBase(ProgramsConfigMixin, CatalogIntegrationMixin, UniqueCourseTest):
"""Base class used for program listing page tests."""
def setUp(self):
super(ProgramPageBase, self).setUp()
self.set_programs_api_configuration(is_enabled=True)
self.programs = [catalog_factories.Program() for __ in range(3)]
self.course_run = catalog_factories.CourseRun(key=self.course_id)
self.stub_catalog_api()
def create_program(self, program_id=None, course_id=None):
"""DRY helper for creating test program data."""
course_id = course_id if course_id else self.course_id
run_mode = program_factories.RunMode(course_key=course_id)
course_code = program_factories.CourseCode(run_modes=[run_mode])
org = program_factories.Organization(key=self.course_info['org'])
if program_id:
program = program_factories.Program(
id=program_id,
status='active',
organizations=[org],
course_codes=[course_code]
)
else:
program = program_factories.Program(
status='active',
organizations=[org],
course_codes=[course_code]
)
return program
def stub_programs_api(self, programs, is_list=True):
"""Stub out the programs API with fake data."""
ProgramsFixture().install_programs(programs, is_list=is_list)
def stub_catalog_api(self):
"""Stub out the catalog API's program and course run endpoints."""
self.set_catalog_configuration(is_enabled=True)
CatalogFixture().install_programs(self.programs)
CatalogFixture().install_course_run(self.course_run)
self.programs = ProgramFactory.create_batch(3)
self.username = None
def auth(self, enroll=True):
"""Authenticate, enrolling the user in the configured course if requested."""
CourseFixture(**self.course_info).install()
course_id = self.course_id if enroll else None
AutoAuthPage(self.browser, course_id=course_id).visit()
auth_page = AutoAuthPage(self.browser, course_id=course_id)
auth_page.visit()
self.username = auth_page.user_info['username']
def create_program(self):
"""DRY helper for creating test program data."""
course_run = CourseRunFactory(key=self.course_id)
course = CourseFactory(course_runs=[course_run])
return ProgramFactory(courses=[course])
def stub_catalog_api(self, data=None):
"""Stub out the catalog API's program and course run endpoints."""
self.set_catalog_integration(is_enabled=True, service_username=self.username)
CatalogFixture().install_programs(data or self.programs)
class ProgramListingPageTest(ProgramPageBase):
......@@ -73,9 +54,8 @@ class ProgramListingPageTest(ProgramPageBase):
def test_no_enrollments(self):
"""Verify that no cards appear when the user has no enrollments."""
program = self.create_program()
self.stub_programs_api([program])
self.auth(enroll=False)
self.stub_catalog_api()
self.listing_page.visit()
......@@ -87,14 +67,8 @@ class ProgramListingPageTest(ProgramPageBase):
Verify that no cards appear when the user has enrollments
but none are included in an active program.
"""
course_id = self.course_id.replace(
self.course_info['run'],
'other_run'
)
program = self.create_program(course_id=course_id)
self.stub_programs_api([program])
self.auth()
self.stub_catalog_api()
self.listing_page.visit()
......@@ -106,10 +80,11 @@ class ProgramListingPageTest(ProgramPageBase):
Verify that cards appear when the user has enrollments
which are included in at least one active program.
"""
program = self.create_program()
self.stub_programs_api([program])
self.auth()
program = self.create_program()
self.stub_catalog_api(data=[program])
self.listing_page.visit()
self.assertTrue(self.listing_page.is_sidebar_present)
......@@ -124,12 +99,13 @@ class ProgramListingPageA11yTest(ProgramPageBase):
self.listing_page = ProgramListingPage(self.browser)
program = self.create_program()
self.stub_programs_api([program])
self.program = self.create_program()
def test_empty_a11y(self):
"""Test a11y of the page's empty state."""
self.auth(enroll=False)
self.stub_catalog_api(data=[self.program])
self.listing_page.visit()
self.assertTrue(self.listing_page.is_sidebar_present)
......@@ -139,6 +115,8 @@ class ProgramListingPageA11yTest(ProgramPageBase):
def test_cards_a11y(self):
"""Test a11y when program cards are present."""
self.auth()
self.stub_catalog_api(data=[self.program])
self.listing_page.visit()
self.assertTrue(self.listing_page.is_sidebar_present)
......@@ -154,11 +132,14 @@ class ProgramDetailsPageA11yTest(ProgramPageBase):
self.details_page = ProgramDetailsPage(self.browser)
program = self.create_program(program_id=self.details_page.program_id)
self.stub_programs_api([program], is_list=False)
self.program = self.create_program()
self.program['uuid'] = self.details_page.program_uuid
def test_a11y(self):
"""Test the page's a11y compliance."""
self.auth()
self.stub_catalog_api(data=self.program)
self.details_page.visit()
self.details_page.a11y_audit.check_for_accessibility_errors()
docs/en_us/internal/testing.rst
......@@ -486,7 +486,7 @@ and the Automated Accessibility Tests `openedx Confluence page
**Prerequisites**:
These prerequisites are all automatically installed and available in `Devstack
<https://github.com/edx/configuration/wiki/edX-Developer-Stack>`__ (since the Cypress release), the supported development enviornment for the edX Platform.
<https://github.com/edx/configuration/wiki/edX-Developer-Stack>`__ (since the Cypress release), the supported development environment for the edX Platform.
* Mongo
......
lms/djangoapps/branding/tests/test_page.py
"""
Tests for branding page
"""
import mock
import datetime
import ddt
from django.conf import settings
from django.contrib.auth.models import AnonymousUser
from django.core.urlresolvers import reverse
from django.http import HttpResponseRedirect
from django.test.utils import override_settings
from django.test.client import RequestFactory
from pytz import UTC
from mock import patch, Mock
from nose.plugins.attrib import attr
from pytz import UTC
from edxmako.shortcuts import render_to_response
from branding.views import index
from courseware.tests.helpers import LoginEnrollmentTestCase
from milestones.tests.utils import MilestonesTestCaseMixin
from util.milestones_helpers import set_prerequisite_courses
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
from django.core.urlresolvers import reverse
from courseware.tests.helpers import LoginEnrollmentTestCase
from util.milestones_helpers import set_prerequisite_courses
from milestones.tests.utils import MilestonesTestCaseMixin
FEATURES_WITH_STARTDATE = settings.FEATURES.copy()
FEATURES_WITH_STARTDATE['DISABLE_START_DATES'] = False
......@@ -290,35 +287,31 @@ class IndexPageCourseCardsSortingTests(ModuleStoreTestCase):
self.assertEqual(context['courses'][2].id, self.course_with_default_start_date.id)
@ddt.ddt
@attr(shard=1)
class IndexPageProgramsTests(ModuleStoreTestCase):
"""
Tests for Programs List in Marketing Pages.
"""
@patch.dict('django.conf.settings.FEATURES', {'DISPLAY_PROGRAMS_ON_MARKETING_PAGES': False})
def test_get_programs_not_called(self):
with mock.patch("student.views.get_programs_data") as patched_get_programs_data:
# check the /dashboard
response = self.client.get('/')
self.assertEqual(response.status_code, 200)
self.assertEqual(patched_get_programs_data.call_count, 0)
with mock.patch("courseware.views.views.get_programs_data") as patched_get_programs_data:
# check the /courses view
response = self.client.get(reverse('branding.views.courses'))
self.assertEqual(response.status_code, 200)
self.assertEqual(patched_get_programs_data.call_count, 0)
@patch.dict('django.conf.settings.FEATURES', {'DISPLAY_PROGRAMS_ON_MARKETING_PAGES': True})
def test_get_programs_called(self):
with mock.patch("student.views.get_programs_data") as patched_get_programs_data:
# check the /dashboard
response = self.client.get('/')
def setUp(self):
super(IndexPageProgramsTests, self).setUp()
self.client.login(username=self.user.username, password=self.user_password)
@ddt.data(True, False)
def test_programs_with_type_logo_called(self, display_programs):
with patch.dict('django.conf.settings.FEATURES', {'DISPLAY_PROGRAMS_ON_MARKETING_PAGES': display_programs}):
views = [
(reverse('dashboard'), 'student.views.get_programs_with_type_logo'),
(reverse('branding.views.courses'), 'courseware.views.views.get_programs_with_type_logo'),
]
for url, dotted_path in views:
with patch(dotted_path) as mock_get_programs_with_type_logo:
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertEqual(patched_get_programs_data.call_count, 1)
with mock.patch("courseware.views.views.get_programs_data") as patched_get_programs_data:
# check the /courses view
response = self.client.get(reverse('branding.views.courses'))
self.assertEqual(response.status_code, 200)
self.assertEqual(patched_get_programs_data.call_count, 1)
if display_programs:
mock_get_programs_with_type_logo.assert_called_once()
else:
mock_get_programs_with_type_logo.assert_not_called_()
lms/djangoapps/courseware/views/views.py
......@@ -40,7 +40,7 @@ from lms.djangoapps.instructor.enrollment import uses_shib
from lms.djangoapps.verify_student.models import SoftwareSecurePhotoVerification
from lms.djangoapps.ccx.custom_exception import CCXLocatorValidationException
from openedx.core.djangoapps.catalog.utils import get_programs_data
from openedx.core.djangoapps.catalog.utils import get_programs_with_type_logo
import shoppingcart
import survey.utils
import survey.views
......@@ -153,7 +153,7 @@ def courses(request):
# for edx-pattern-library is added.
if configuration_helpers.get_value("DISPLAY_PROGRAMS_ON_MARKETING_PAGES",
settings.FEATURES.get("DISPLAY_PROGRAMS_ON_MARKETING_PAGES")):
programs_list = get_programs_data(request.user)
programs_list = get_programs_with_type_logo()
return render_to_response(
"courseware/courses.html",
......
lms/djangoapps/learner_dashboard/tests/test_programs.py
......@@ -6,36 +6,34 @@ import json
import re
import unittest
from urlparse import urljoin
from uuid import uuid4
from bs4 import BeautifulSoup
from django.conf import settings
from django.core.urlresolvers import reverse
from django.test import override_settings
from django.utils.text import slugify
from edx_oauth2_provider.tests.factories import ClientFactory
import httpretty
import mock
from provider.constants import CONFIDENTIAL
from openedx.core.djangoapps.credentials.models import CredentialsApiConfig
from openedx.core.djangoapps.credentials.tests import factories as credentials_factories
from openedx.core.djangoapps.catalog.tests.factories import ProgramFactory, CourseFactory, CourseRunFactory
from openedx.core.djangoapps.catalog.tests.mixins import CatalogIntegrationMixin
from openedx.core.djangoapps.catalog.utils import munge_catalog_program
from openedx.core.djangoapps.credentials.tests.factories import UserCredential, ProgramCredential
from openedx.core.djangoapps.credentials.tests.mixins import CredentialsApiConfigMixin
from openedx.core.djangoapps.programs.models import ProgramsApiConfig
from openedx.core.djangoapps.programs.tests import factories as programs_factories
from openedx.core.djangoapps.programs.tests.mixins import ProgramsApiConfigMixin
from openedx.core.djangolib.testing.utils import skip_unless_lms, toggle_switch
from openedx.core.djangolib.testing.utils import skip_unless_lms
from student.tests.factories import UserFactory, CourseEnrollmentFactory
from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
from xmodule.modulestore.tests.factories import CourseFactory as ModuleStoreCourseFactory
UTILS_MODULE = 'openedx.core.djangoapps.programs.utils'
MARKETING_URL = 'https://www.example.com/marketing/path'
CATALOG_UTILS_MODULE = 'openedx.core.djangoapps.catalog.utils'
CREDENTIALS_UTILS_MODULE = 'openedx.core.djangoapps.credentials.utils'
PROGRAMS_UTILS_MODULE = 'openedx.core.djangoapps.programs.utils'
@skip_unless_lms
@httpretty.activate
@override_settings(MKTG_URLS={'ROOT': 'https://www.example.com'})
@mock.patch(PROGRAMS_UTILS_MODULE + '.get_programs')
class TestProgramListing(ProgramsApiConfigMixin, CredentialsApiConfigMixin, SharedModuleStoreTestCase):
"""Unit tests for the program listing page."""
maxDiff = None
......@@ -46,22 +44,12 @@ class TestProgramListing(ProgramsApiConfigMixin, CredentialsApiConfigMixin, Shar
def setUpClass(cls):
super(TestProgramListing, cls).setUpClass()
for name in [ProgramsApiConfig.OAUTH2_CLIENT_NAME, CredentialsApiConfig.OAUTH2_CLIENT_NAME]:
ClientFactory(name=name, client_type=CONFIDENTIAL)
cls.course = ModuleStoreCourseFactory()
course_run = CourseRunFactory(key=unicode(cls.course.id)) # pylint: disable=no-member
course = CourseFactory(course_runs=[course_run])
cls.course = CourseFactory()
organization = programs_factories.Organization()
run_mode = programs_factories.RunMode(course_key=unicode(cls.course.id)) # pylint: disable=no-member
course_code = programs_factories.CourseCode(run_modes=[run_mode])
cls.first_program = programs_factories.Program(
organizations=[organization],
course_codes=[course_code]
)
cls.second_program = programs_factories.Program(
organizations=[organization],
course_codes=[course_code]
)
cls.first_program = ProgramFactory(courses=[course])
cls.second_program = ProgramFactory(courses=[course])
cls.data = sorted([cls.first_program, cls.second_program], key=cls.program_sort_key)
......@@ -76,7 +64,13 @@ class TestProgramListing(ProgramsApiConfigMixin, CredentialsApiConfigMixin, Shar
"""
Helper function used to sort dictionaries representing programs.
"""
return program['id']
try:
return program['title']
except: # pylint: disable=bare-except
# This is here temporarily because programs are still being munged
# to look like they came from the programs service before going out
# to the front end.
return program['name']
def credential_sort_key(self, credential):
"""
......@@ -87,27 +81,6 @@ class TestProgramListing(ProgramsApiConfigMixin, CredentialsApiConfigMixin, Shar
except KeyError:
return credential['credential_url']
def mock_programs_api(self, data):
"""Helper for mocking out Programs API URLs."""
self.assertTrue(httpretty.is_enabled(), msg='httpretty must be enabled to mock Programs API calls.')
url = ProgramsApiConfig.current().internal_api_url.strip('/') + '/programs/'
body = json.dumps({'results': data})
httpretty.register_uri(httpretty.GET, url, body=body, content_type='application/json')
def mock_credentials_api(self, data):
"""Helper for mocking out Credentials API URLs."""
self.assertTrue(httpretty.is_enabled(), msg='httpretty must be enabled to mock Credentials API calls.')
url = '{base}/credentials/?username={username}'.format(
base=CredentialsApiConfig.current().internal_api_url.strip('/'),
username=self.user.username
)
body = json.dumps({'results': data})
httpretty.register_uri(httpretty.GET, url, body=body, content_type='application/json')
def load_serialized_data(self, response, key):
"""
Extract and deserialize serialized data from the response.
......@@ -131,12 +104,12 @@ class TestProgramListing(ProgramsApiConfigMixin, CredentialsApiConfigMixin, Shar
self.assertEqual(subset, intersection)
def test_login_required(self):
def test_login_required(self, mock_get_programs):
"""
Verify that login is required to access the page.
"""
self.create_programs_config()
self.mock_programs_api(self.data)
mock_get_programs.return_value = self.data
self.client.logout()
......@@ -151,7 +124,7 @@ class TestProgramListing(ProgramsApiConfigMixin, CredentialsApiConfigMixin, Shar
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
def test_404_if_disabled(self):
def test_404_if_disabled(self, _mock_get_programs):
"""
Verify that the page 404s if disabled.
"""
......@@ -160,22 +133,22 @@ class TestProgramListing(ProgramsApiConfigMixin, CredentialsApiConfigMixin, Shar
response = self.client.get(self.url)
self.assertEqual(response.status_code, 404)
def test_empty_state(self):
def test_empty_state(self, mock_get_programs):
"""
Verify that the response contains no programs data when no programs are engaged.
"""
self.create_programs_config()
self.mock_programs_api(self.data)
mock_get_programs.return_value = self.data
response = self.client.get(self.url)
self.assertContains(response, 'programsData: []')
def test_programs_listed(self):
def test_programs_listed(self, mock_get_programs):
"""
Verify that the response contains accurate programs data when programs are engaged.
"""
self.create_programs_config()
self.mock_programs_api(self.data)
mock_get_programs.return_value = self.data
CourseEnrollmentFactory(user=self.user, course_id=self.course.id) # pylint: disable=no-member
......@@ -184,27 +157,27 @@ class TestProgramListing(ProgramsApiConfigMixin, CredentialsApiConfigMixin, Shar
actual = sorted(actual, key=self.program_sort_key)
for index, actual_program in enumerate(actual):
expected_program = self.data[index]
expected_program = munge_catalog_program(self.data[index])
self.assert_dict_contains_subset(actual_program, expected_program)
def test_program_discovery(self):
def test_program_discovery(self, mock_get_programs):
"""
Verify that a link to a programs marketing page appears in the response.
"""
self.create_programs_config(marketing_path='bar')
self.mock_programs_api(self.data)
mock_get_programs.return_value = self.data
marketing_root = urljoin(settings.MKTG_URLS.get('ROOT'), 'bar').rstrip('/')
response = self.client.get(self.url)
self.assertContains(response, marketing_root)
def test_links_to_detail_pages(self):
def test_links_to_detail_pages(self, mock_get_programs):
"""
Verify that links to detail pages are present.
"""
self.create_programs_config()
self.mock_programs_api(self.data)
mock_get_programs.return_value = self.data
CourseEnrollmentFactory(user=self.user, course_id=self.course.id) # pylint: disable=no-member
......@@ -215,43 +188,41 @@ class TestProgramListing(ProgramsApiConfigMixin, CredentialsApiConfigMixin, Shar
for index, actual_program in enumerate(actual):
expected_program = self.data[index]
base = reverse('program_details_view', args=[expected_program['id']]).rstrip('/')
slug = slugify(expected_program['name'])
self.assertEqual(
actual_program['detail_url'],
'{}/{}'.format(base, slug)
)
expected_url = reverse('program_details_view', kwargs={'program_uuid': expected_program['uuid']})
self.assertEqual(actual_program['detail_url'], expected_url)
def test_certificates_listed(self):
@mock.patch(CREDENTIALS_UTILS_MODULE + '.get_user_credentials')
@mock.patch(CREDENTIALS_UTILS_MODULE + '.get_programs')
def test_certificates_listed(self, mock_get_programs, mock_get_user_credentials, __):
"""
Verify that the response contains accurate certificate data when certificates are available.
"""
self.create_programs_config()
self.create_credentials_config(is_learner_issuance_enabled=True)
self.mock_programs_api(self.data)
mock_get_programs.return_value = self.data
first_credential = credentials_factories.UserCredential(
first_credential = UserCredential(
username=self.user.username,
credential=credentials_factories.ProgramCredential(
program_id=self.first_program['id']
credential=ProgramCredential(
program_uuid=self.first_program['uuid']
)
)
second_credential = credentials_factories.UserCredential(
second_credential = UserCredential(
username=self.user.username,
credential=credentials_factories.ProgramCredential(
program_id=self.second_program['id']
credential=ProgramCredential(
program_uuid=self.second_program['uuid']
)
)
credentials_data = sorted([first_credential, second_credential], key=self.credential_sort_key)
self.mock_credentials_api(credentials_data)
mock_get_user_credentials.return_value = credentials_data
response = self.client.get(self.url)
actual = self.load_serialized_data(response, 'certificatesData')
actual = sorted(actual, key=self.credential_sort_key)
self.assertEqual(len(actual), len(credentials_data))
for index, actual_credential in enumerate(actual):
expected_credential = credentials_data[index]
......@@ -262,51 +233,24 @@ class TestProgramListing(ProgramsApiConfigMixin, CredentialsApiConfigMixin, Shar
expected_credential['certificate_url']
)
def test_switch_to_catalog(self):
"""
Verify that the 'get_programs_from_catalog' switch can be used to route
traffic between the programs and catalog services.
"""
self.create_programs_config()
switch_name = 'get_programs_from_catalog'
with mock.patch('openedx.core.djangoapps.programs.utils.get_programs') as mock_get_programs:
mock_get_programs.return_value = self.data
toggle_switch(switch_name)
self.client.get(self.url)
mock_get_programs.assert_called_with(self.user, use_catalog=True)
toggle_switch(switch_name)
self.client.get(self.url)
mock_get_programs.assert_called_with(self.user, use_catalog=False)
@skip_unless_lms
@httpretty.activate
@override_settings(MKTG_URLS={'ROOT': 'https://www.example.com'})
@mock.patch(UTILS_MODULE + '.get_run_marketing_url', mock.Mock(return_value=MARKETING_URL))
class TestProgramDetails(ProgramsApiConfigMixin, SharedModuleStoreTestCase):
@mock.patch(CATALOG_UTILS_MODULE + '.get_edx_api_data')
class TestProgramDetails(ProgramsApiConfigMixin, CatalogIntegrationMixin, SharedModuleStoreTestCase):
"""Unit tests for the program details page."""
program_id = 123
program_uuid = str(uuid4())
password = 'test'
url = reverse('program_details_view', args=[program_id])
url = reverse('program_details_view', kwargs={'program_uuid': program_uuid})
@classmethod
def setUpClass(cls):
super(TestProgramDetails, cls).setUpClass()
ClientFactory(name=ProgramsApiConfig.OAUTH2_CLIENT_NAME, client_type=CONFIDENTIAL)
modulestore_course = ModuleStoreCourseFactory()
course_run = CourseRunFactory(key=unicode(modulestore_course.id)) # pylint: disable=no-member
course = CourseFactory(course_runs=[course_run])
course = CourseFactory()
organization = programs_factories.Organization()
run_mode = programs_factories.RunMode(course_key=unicode(course.id)) # pylint: disable=no-member
course_code = programs_factories.CourseCode(run_modes=[run_mode])
cls.data = programs_factories.Program(
organizations=[organization],
course_codes=[course_code]
)
cls.data = ProgramFactory(uuid=cls.program_uuid, courses=[course])
def setUp(self):
super(TestProgramDetails, self).setUp()
......@@ -314,31 +258,12 @@ class TestProgramDetails(ProgramsApiConfigMixin, SharedModuleStoreTestCase):
self.user = UserFactory()
self.client.login(username=self.user.username, password=self.password)
def mock_programs_api(self, data, status=200):
"""Helper for mocking out Programs API URLs."""
self.assertTrue(httpretty.is_enabled(), msg='httpretty must be enabled to mock Programs API calls.')
url = '{api_root}/programs/{id}/'.format(
api_root=ProgramsApiConfig.current().internal_api_url.strip('/'),
id=self.program_id
)
body = json.dumps(data)
httpretty.register_uri(
httpretty.GET,
url,
body=body,
status=status,
content_type='application/json',
)
def assert_program_data_present(self, response):
"""Verify that program data is present."""
self.assertContains(response, 'programData')
self.assertContains(response, 'urls')
self.assertContains(response, 'program_listing_url')
self.assertContains(response, self.data['name'])
self.assertContains(response, self.data['title'])
self.assert_programs_tab_present(response)
def assert_programs_tab_present(self, response):
......@@ -348,12 +273,16 @@ class TestProgramDetails(ProgramsApiConfigMixin, SharedModuleStoreTestCase):
any(soup.find_all('a', class_='tab-nav-link', href=reverse('program_listing_view')))
)
def test_login_required(self):
def test_login_required(self, mock_get_edx_api_data):
"""
Verify that login is required to access the page.
"""
self.create_programs_config()
self.mock_programs_api(self.data)
catalog_integration = self.create_catalog_integration()
UserFactory(username=catalog_integration.service_username)
mock_get_edx_api_data.return_value = self.data
self.client.logout()
......@@ -368,7 +297,7 @@ class TestProgramDetails(ProgramsApiConfigMixin, SharedModuleStoreTestCase):
response = self.client.get(self.url)
self.assert_program_data_present(response)
def test_404_if_disabled(self):
def test_404_if_disabled(self, _mock_get_edx_api_data):
"""
Verify that the page 404s if disabled.
"""
......@@ -377,30 +306,9 @@ class TestProgramDetails(ProgramsApiConfigMixin, SharedModuleStoreTestCase):
response = self.client.get(self.url)
self.assertEqual(response.status_code, 404)
def test_404_if_no_data(self):
def test_404_if_no_data(self, _mock_get_edx_api_data):
"""Verify that the page 404s if no program data is found."""
self.create_programs_config()
self.mock_programs_api(self.data, status=404)
response = self.client.get(self.url)
self.assertEqual(response.status_code, 404)
httpretty.reset()
self.mock_programs_api({})
response = self.client.get(self.url)
self.assertEqual(response.status_code, 404)
def test_page_routing(self):
"""Verify that the page can be hit with or without a program name in the URL."""
self.create_programs_config()
self.mock_programs_api(self.data)
response = self.client.get(self.url)
self.assert_program_data_present(response)
response = self.client.get(self.url + 'program_name/')
self.assert_program_data_present(response)
response = self.client.get(self.url + 'program_name/invalid/')
self.assertEqual(response.status_code, 404)
lms/djangoapps/learner_dashboard/urls.py
......@@ -6,7 +6,5 @@ from . import views
urlpatterns = [
url(r'^programs/$', views.program_listing, name='program_listing_view'),
# Matches paths like 'programs/123/' and 'programs/123/foo/', but not 'programs/123/foo/bar/'.
# Also accepts strings that look like UUIDs, to support retrieval of catalog-based MicroMasters.
url(r'^programs/(?P<program_id>[0-9a-f-]+)/[\w\-]*/?$', views.program_details, name='program_details_view'),
url(r'^programs/(?P<program_uuid>[0-9a-f-]+)/$', views.program_details, name='program_details_view'),
]
lms/djangoapps/learner_dashboard/views.py
"""Learner dashboard views"""
import uuid
from django.contrib.auth.decorators import login_required
from django.core.urlresolvers import reverse
from django.http import Http404
......@@ -8,12 +6,16 @@ from django.views.decorators.http import require_GET
from edxmako.shortcuts import render_to_response
from lms.djangoapps.learner_dashboard.utils import strip_course_id, FAKE_COURSE_KEY
from openedx.core.djangoapps.catalog.utils import get_programs as get_catalog_programs, munge_catalog_program
from openedx.core.djangoapps.catalog.utils import get_programs, munge_catalog_program
from openedx.core.djangoapps.credentials.utils import get_programs_credentials
from openedx.core.djangoapps.programs.models import ProgramsApiConfig
from openedx.core.djangoapps.programs import utils
from openedx.core.djangoapps.programs.utils import (
get_program_marketing_url,
munge_progress_map,
ProgramProgressMeter,
ProgramDataExtender,
)
from openedx.core.djangoapps.user_api.preferences.api import get_user_preferences
import waffle
@login_required
......@@ -24,16 +26,17 @@ def program_listing(request):
if not programs_config.show_program_listing:
raise Http404
use_catalog = waffle.switch_is_active('get_programs_from_catalog')
meter = utils.ProgramProgressMeter(request.user, use_catalog=use_catalog)
meter = ProgramProgressMeter(request.user)
engaged_programs = [munge_catalog_program(program) for program in meter.engaged_programs]
progress = [munge_progress_map(progress_map) for progress_map in meter.progress]
context = {
'credentials': get_programs_credentials(request.user),
'disable_courseware_js': True,
'marketing_url': utils.get_program_marketing_url(programs_config),
'marketing_url': get_program_marketing_url(programs_config),
'nav_hidden': True,
'programs': meter.engaged_programs(),
'progress': meter.progress,
'programs': engaged_programs,
'progress': progress,
'show_program_listing': programs_config.show_program_listing,
'uses_pattern_library': True,
}
......@@ -43,26 +46,18 @@ def program_listing(request):
@login_required
@require_GET
def program_details(request, program_id):
def program_details(request, program_uuid):
"""View details about a specific program."""
programs_config = ProgramsApiConfig.current()
if not programs_config.show_program_details:
raise Http404
try:
# If the ID is a UUID, the requested program resides in the catalog.
uuid.UUID(program_id)
program_data = get_catalog_programs(request.user, uuid=program_id)
if program_data:
program_data = munge_catalog_program(program_data)
except ValueError:
program_data = utils.get_programs(request.user, program_id=program_id)
program_data = get_programs(uuid=program_uuid)
if not program_data:
raise Http404
program_data = utils.ProgramDataExtender(program_data, request.user).extend()
program_data = munge_catalog_program(program_data)
program_data = ProgramDataExtender(program_data, request.user).extend()
urls = {
'program_listing_url': reverse('program_listing_view'),
......
openedx/core/djangoapps/catalog/tests/factories.py
"""Factories for generating fake catalog data."""
# pylint: disable=missing-docstring, invalid-name
from random import randint
from functools import partial
import factory
from faker import Faker
......@@ -9,6 +9,14 @@ from faker import Faker
fake = Faker()
def generate_instances(factory_class, count=3):
"""
Use this to populate fields with values derived from other factories. If
the array is used directly, the same value will be used repeatedly.
"""
return factory_class.create_batch(count)
def generate_course_key():
return '+'.join(fake.words(2))
......@@ -26,17 +34,17 @@ def generate_zulu_datetime():
return fake.date_time().isoformat() + 'Z'
class DictFactory(factory.Factory):
class DictFactoryBase(factory.Factory):
class Meta(object):
model = dict
class ImageFactory(DictFactory):
class ImageFactoryBase(DictFactoryBase):
height = factory.Faker('random_int')
width = factory.Faker('random_int')
class Image(ImageFactory):
class ImageFactory(ImageFactoryBase):
"""
For constructing dicts mirroring the catalog's serialized representation of ImageFields.
......@@ -46,7 +54,7 @@ class Image(ImageFactory):
src = factory.Faker('image_url')
class StdImage(ImageFactory):
class StdImageFactory(ImageFactoryBase):
"""
For constructing dicts mirroring the catalog's serialized representation of StdImageFields.
......@@ -57,21 +65,21 @@ class StdImage(ImageFactory):
def generate_sized_stdimage():
return {
size: StdImage() for size in ['large', 'medium', 'small', 'x-small']
size: StdImageFactory() for size in ['large', 'medium', 'small', 'x-small']
}
class Organization(DictFactory):
class OrganizationFactory(DictFactoryBase):
key = factory.Faker('word')
name = factory.Faker('company')
uuid = factory.Faker('uuid4')
class CourseRun(DictFactory):
class CourseRunFactory(DictFactoryBase):
end = factory.LazyFunction(generate_zulu_datetime)
enrollment_end = factory.LazyFunction(generate_zulu_datetime)
enrollment_start = factory.LazyFunction(generate_zulu_datetime)
image = Image()
image = ImageFactory()
key = factory.LazyFunction(generate_course_run_key)
marketing_url = factory.Faker('url')
pacing_type = 'self_paced'
......@@ -82,20 +90,20 @@ class CourseRun(DictFactory):
uuid = factory.Faker('uuid4')
class Course(DictFactory):
course_runs = [CourseRun() for __ in range(randint(3, 5))]
image = Image()
class CourseFactory(DictFactoryBase):
course_runs = factory.LazyFunction(partial(generate_instances, CourseRunFactory))
image = ImageFactory()
key = factory.LazyFunction(generate_course_key)
owners = [Organization()]
owners = factory.LazyFunction(partial(generate_instances, OrganizationFactory, count=1))
title = factory.Faker('catch_phrase')
uuid = factory.Faker('uuid4')
class Program(DictFactory):
authoring_organizations = [Organization()]
class ProgramFactory(DictFactoryBase):
authoring_organizations = factory.LazyFunction(partial(generate_instances, OrganizationFactory, count=1))
banner_image = factory.LazyFunction(generate_sized_stdimage)
card_image_url = factory.Faker('image_url')
courses = [Course() for __ in range(randint(3, 5))]
courses = factory.LazyFunction(partial(generate_instances, CourseFactory))
marketing_slug = factory.Faker('slug')
marketing_url = factory.Faker('url')
status = 'active'
......@@ -105,6 +113,6 @@ class Program(DictFactory):
uuid = factory.Faker('uuid4')
class ProgramType(DictFactory):
class ProgramTypeFactory(DictFactoryBase):
name = factory.Faker('word')
logo_image = factory.LazyFunction(generate_sized_stdimage)
openedx/core/djangoapps/catalog/tests/mixins.py
......@@ -5,16 +5,19 @@ from openedx.core.djangoapps.catalog.models import CatalogIntegration
class CatalogIntegrationMixin(object):
"""Utility for working with the catalog service during testing."""
DEFAULTS = {
catalog_integration_defaults = {
'enabled': True,
'internal_api_url': 'https://catalog-internal.example.com/api/v1/',
'cache_ttl': 0,
'service_username': 'lms_catalog_service_user'
'service_username': 'lms_catalog_service_user',
}
def create_catalog_integration(self, **kwargs):
"""Creates a new CatalogIntegration with DEFAULTS, updated with any provided overrides."""
fields = dict(self.DEFAULTS, **kwargs)
"""
Creates a new CatalogIntegration with catalog_integration_defaults,
updated with any provided overrides.
"""
fields = dict(self.catalog_integration_defaults, **kwargs)
CatalogIntegration(**fields).save()
return CatalogIntegration.current()
openedx/core/djangoapps/catalog/tests/test_utils.py
"""Tests covering utilities for integrating with the catalog service."""
# pylint: disable=missing-docstring
import uuid
import copy
from django.contrib.auth import get_user_model
from django.test import TestCase
import mock
from opaque_keys.edx.keys import CourseKey
from openedx.core.djangoapps.catalog import utils
from openedx.core.djangoapps.catalog.models import CatalogIntegration
from openedx.core.djangoapps.catalog.tests import factories, mixins
from student.tests.factories import UserFactory, AnonymousUserFactory
from openedx.core.djangoapps.catalog.tests.factories import ProgramFactory, ProgramTypeFactory
from openedx.core.djangoapps.catalog.tests.mixins import CatalogIntegrationMixin
from openedx.core.djangoapps.catalog.utils import (
get_programs,
munge_catalog_program,
get_program_types,
get_programs_with_type_logo,
)
from openedx.core.djangolib.testing.utils import skip_unless_lms
from student.tests.factories import UserFactory
UTILS_MODULE = 'openedx.core.djangoapps.catalog.utils'
User = get_user_model() # pylint: disable=invalid-name
@skip_unless_lms
@mock.patch(UTILS_MODULE + '.get_edx_api_data')
# ConfigurationModels use the cache. Make every cache get a miss.
@mock.patch('config_models.models.cache.get', return_value=None)
class TestGetPrograms(mixins.CatalogIntegrationMixin, TestCase):
class TestGetPrograms(CatalogIntegrationMixin, TestCase):
"""Tests covering retrieval of programs from the catalog service."""
def setUp(self):
super(TestGetPrograms, self).setUp()
self.user = UserFactory()
self.uuid = str(uuid.uuid4())
self.type = 'FooBar'
self.catalog_integration = self.create_catalog_integration(cache_ttl=1)
UserFactory(username=self.catalog_integration.service_username)
def assert_contract(self, call_args, program_uuid=None, type=None): # pylint: disable=redefined-builtin
"""Verify that API data retrieval utility is used correctly."""
args, kwargs = call_args
for arg in (self.catalog_integration, self.user, 'programs'):
for arg in (self.catalog_integration, 'programs'):
self.assertIn(arg, args)
self.assertEqual(kwargs['resource_id'], program_uuid)
......@@ -57,138 +68,88 @@ class TestGetPrograms(mixins.CatalogIntegrationMixin, TestCase):
return args, kwargs
def test_get_programs(self, _mock_cache, mock_get_catalog_data):
programs = [factories.Program() for __ in range(3)]
mock_get_catalog_data.return_value = programs
data = utils.get_programs(self.user)
self.assert_contract(mock_get_catalog_data.call_args)
self.assertEqual(data, programs)
def test_get_programs_anonymous_user(self, _mock_cache, mock_get_catalog_data):
programs = [factories.Program() for __ in range(3)]
mock_get_catalog_data.return_value = programs
anonymous_user = AnonymousUserFactory()
def test_get_programs(self, mock_get_edx_api_data):
programs = [ProgramFactory() for __ in range(3)]
mock_get_edx_api_data.return_value = programs
# The user is an Anonymous user but the Catalog Service User has not been created yet.
data = utils.get_programs(anonymous_user)
# This should not return programs.
self.assertEqual(data, [])
data = get_programs()
UserFactory(username='lms_catalog_service_user')
# After creating the service user above,
data = utils.get_programs(anonymous_user)
# the programs should be returned successfully.
self.assert_contract(mock_get_edx_api_data.call_args)
self.assertEqual(data, programs)
def test_get_program_types(self, _mock_cache, mock_get_catalog_data):
program_types = [factories.ProgramType() for __ in range(3)]
mock_get_catalog_data.return_value = program_types
# Creating Anonymous user but the Catalog Service User has not been created yet.
anonymous_user = AnonymousUserFactory()
data = utils.get_program_types(anonymous_user)
# This should not return programs.
self.assertEqual(data, [])
# Creating Catalog Service User user
UserFactory(username='lms_catalog_service_user')
data = utils.get_program_types(anonymous_user)
# the programs should be returned successfully.
self.assertEqual(data, program_types)
# Catalog integration is disabled now.
self.catalog_integration = self.create_catalog_integration(enabled=False)
data = utils.get_program_types(anonymous_user)
# This should not return programs.
self.assertEqual(data, [])
def test_get_programs_data(self, _mock_cache, mock_get_catalog_data): # pylint: disable=unused-argument
programs = []
program_types = []
programs_data = []
def test_get_one_program(self, mock_get_edx_api_data):
program = ProgramFactory()
mock_get_edx_api_data.return_value = program
for index in range(3):
# Creating the Programs and their corresponding program types.
type_name = "type_name_{postfix}".format(postfix=index)
program = factories.Program(type=type_name)
program_type = factories.ProgramType(name=type_name)
data = get_programs(uuid=self.uuid)
# Maintaining the programs, program types and program data(program+logo_image) lists.
programs.append(program)
program_types.append(program_type)
programs_data.append(copy.deepcopy(program))
# Adding the logo image in program data.
programs_data[-1]['logo_image'] = program_type["logo_image"]
with mock.patch("openedx.core.djangoapps.catalog.utils.get_programs") as patched_get_programs:
with mock.patch("openedx.core.djangoapps.catalog.utils.get_program_types") as patched_get_program_types:
# Mocked the "get_programs" and "get_program_types"
patched_get_programs.return_value = programs
patched_get_program_types.return_value = program_types
programs_data = utils.get_programs_data()
self.assertEqual(programs_data, programs)
def test_get_one_program(self, _mock_cache, mock_get_catalog_data):
program = factories.Program()
mock_get_catalog_data.return_value = program
data = utils.get_programs(self.user, uuid=self.uuid)
self.assert_contract(mock_get_catalog_data.call_args, program_uuid=self.uuid)
self.assert_contract(mock_get_edx_api_data.call_args, program_uuid=self.uuid)
self.assertEqual(data, program)
def test_get_programs_by_type(self, _mock_cache, mock_get_catalog_data):
programs = [factories.Program() for __ in range(2)]
mock_get_catalog_data.return_value = programs
def test_get_programs_by_type(self, mock_get_edx_api_data):
programs = ProgramFactory.create_batch(2)
mock_get_edx_api_data.return_value = programs
data = utils.get_programs(self.user, type=self.type)
data = get_programs(type=self.type)
self.assert_contract(mock_get_catalog_data.call_args, type=self.type)
self.assert_contract(mock_get_edx_api_data.call_args, type=self.type)
self.assertEqual(data, programs)
def test_programs_unavailable(self, _mock_cache, mock_get_catalog_data):
mock_get_catalog_data.return_value = []
def test_programs_unavailable(self, mock_get_edx_api_data):
mock_get_edx_api_data.return_value = []
data = utils.get_programs(self.user)
data = get_programs()
self.assert_contract(mock_get_catalog_data.call_args)
self.assert_contract(mock_get_edx_api_data.call_args)
self.assertEqual(data, [])
def test_cache_disabled(self, _mock_cache, mock_get_catalog_data):
def test_cache_disabled(self, mock_get_edx_api_data):
self.catalog_integration = self.create_catalog_integration(cache_ttl=0)
utils.get_programs(self.user)
self.assert_contract(mock_get_catalog_data.call_args)
def test_config_missing(self, _mock_cache, _mock_get_catalog_data):
"""Verify that no errors occur if this method is called when catalog config is missing."""
get_programs()
self.assert_contract(mock_get_edx_api_data.call_args)
def test_config_missing(self, _mock_get_edx_api_data):
"""
Verify that no errors occur if this method is called when catalog config
is missing.
"""
CatalogIntegration.objects.all().delete()
data = utils.get_programs(self.user)
data = get_programs()
self.assertEqual(data, [])
def test_service_user_missing(self, _mock_get_edx_api_data):
"""
Verify that no errors occur if this method is called when the catalog
service user is missing.
"""
# Note: Deleting the service user would be ideal, but causes mysterious
# errors on Jenkins.
self.create_catalog_integration(service_username='nonexistent-user')
data = get_programs()
self.assertEqual(data, [])
class TestMungeCatalogProgram(TestCase):
"""Tests covering querystring stripping."""
catalog_program = factories.Program()
def setUp(self):
super(TestMungeCatalogProgram, self).setUp()
def test_munge_catalog_program(self):
munged = utils.munge_catalog_program(self.catalog_program)
self.catalog_program = ProgramFactory()
def assert_munged(self, program):
munged = munge_catalog_program(program)
expected = {
'id': self.catalog_program['uuid'],
'name': self.catalog_program['title'],
'subtitle': self.catalog_program['subtitle'],
'category': self.catalog_program['type'],
'marketing_slug': self.catalog_program['marketing_slug'],
'id': program['uuid'],
'name': program['title'],
'subtitle': program['subtitle'],
'category': program['type'],
'marketing_slug': program['marketing_slug'],
'organizations': [
{
'display_name': organization['name'],
'key': organization['key']
} for organization in self.catalog_program['authoring_organizations']
} for organization in program['authoring_organizations']
],
'course_codes': [
{
......@@ -200,108 +161,72 @@ class TestMungeCatalogProgram(TestCase):
},
'run_modes': [
{
'course_key': run['key'],
'run_key': CourseKey.from_string(run['key']).run,
'mode_slug': 'verified'
} for run in course['course_runs']
'course_key': course_run['key'],
'run_key': CourseKey.from_string(course_run['key']).run,
'mode_slug': course_run['type'],
'marketing_url': course_run['marketing_url'],
} for course_run in course['course_runs']
],
} for course in self.catalog_program['courses']
} for course in program['courses']
],
'banner_image_urls': {
'w1440h480': self.catalog_program['banner_image']['large']['url'],
'w726h242': self.catalog_program['banner_image']['medium']['url'],
'w435h145': self.catalog_program['banner_image']['small']['url'],
'w348h116': self.catalog_program['banner_image']['x-small']['url'],
'w1440h480': program['banner_image']['large']['url'],
'w726h242': program['banner_image']['medium']['url'],
'w435h145': program['banner_image']['small']['url'],
'w348h116': program['banner_image']['x-small']['url'],
},
'detail_url': program.get('detail_url'),
}
self.assertEqual(munged, expected)
def test_munge_catalog_program(self):
self.assert_munged(self.catalog_program)
@mock.patch(UTILS_MODULE + '.get_edx_api_data')
@mock.patch('config_models.models.cache.get', return_value=None)
class TestGetCourseRun(mixins.CatalogIntegrationMixin, TestCase):
"""Tests covering retrieval of course runs from the catalog service."""
def setUp(self):
super(TestGetCourseRun, self).setUp()
self.user = UserFactory()
self.course_key = CourseKey.from_string('foo/bar/baz')
self.catalog_integration = self.create_catalog_integration()
def assert_contract(self, call_args):
"""Verify that API data retrieval utility is used correctly."""
args, kwargs = call_args
for arg in (self.catalog_integration, self.user, 'course_runs'):
self.assertIn(arg, args)
self.assertEqual(kwargs['resource_id'], unicode(self.course_key))
self.assertEqual(kwargs['api']._store['base_url'], self.catalog_integration.internal_api_url) # pylint: disable=protected-access
return args, kwargs
def test_get_course_run(self, _mock_cache, mock_get_catalog_data):
course_run = factories.CourseRun()
mock_get_catalog_data.return_value = course_run
data = utils.get_course_run(self.course_key, self.user)
self.assert_contract(mock_get_catalog_data.call_args)
self.assertEqual(data, course_run)
def test_course_run_unavailable(self, _mock_cache, mock_get_catalog_data):
mock_get_catalog_data.return_value = []
data = utils.get_course_run(self.course_key, self.user)
self.assert_contract(mock_get_catalog_data.call_args)
self.assertEqual(data, {})
def test_cache_disabled(self, _mock_cache, mock_get_catalog_data):
utils.get_course_run(self.course_key, self.user)
_, kwargs = self.assert_contract(mock_get_catalog_data.call_args)
self.assertIsNone(kwargs['cache_key'])
def test_cache_enabled(self, _mock_cache, mock_get_catalog_data):
catalog_integration = self.create_catalog_integration(cache_ttl=1)
utils.get_course_run(self.course_key, self.user)
_, kwargs = mock_get_catalog_data.call_args
self.assertEqual(kwargs['cache_key'], catalog_integration.CACHE_KEY)
def test_config_missing(self, _mock_cache, _mock_get_catalog_data):
"""Verify that no errors occur if this method is called when catalog config is missing."""
CatalogIntegration.objects.all().delete()
data = utils.get_course_run(self.course_key, self.user)
self.assertEqual(data, {})
def test_munge_with_detail_url(self):
self.catalog_program['detail_url'] = 'foo'
self.assert_munged(self.catalog_program)
@mock.patch(UTILS_MODULE + '.get_course_run')
class TestGetRunMarketingUrl(TestCase):
"""Tests covering retrieval of course run marketing URLs."""
def setUp(self):
super(TestGetRunMarketingUrl, self).setUp()
@skip_unless_lms
@mock.patch(UTILS_MODULE + '.get_edx_api_data')
class TestGetProgramTypes(CatalogIntegrationMixin, TestCase):
"""Tests covering retrieval of program types from the catalog service."""
def test_get_program_types(self, mock_get_edx_api_data):
program_types = [ProgramTypeFactory() for __ in range(3)]
mock_get_edx_api_data.return_value = program_types
# Catalog integration is disabled.
data = get_program_types()
self.assertEqual(data, [])
self.course_key = CourseKey.from_string('foo/bar/baz')
self.user = UserFactory()
catalog_integration = self.create_catalog_integration()
UserFactory(username=catalog_integration.service_username)
data = get_program_types()
self.assertEqual(data, program_types)
def test_get_run_marketing_url(self, mock_get_course_run):
course_run = factories.CourseRun()
mock_get_course_run.return_value = course_run
def test_get_programs_with_type_logo(self, _mock_get_edx_api_data):
programs = []
program_types = []
programs_with_type_logo = []
url = utils.get_run_marketing_url(self.course_key, self.user)
for index in range(3):
# Creating the Programs and their corresponding program types.
type_name = 'type_name_{postfix}'.format(postfix=index)
program = ProgramFactory(type=type_name)
program_type = ProgramTypeFactory(name=type_name)
self.assertEqual(url, course_run['marketing_url'])
programs.append(program)
program_types.append(program_type)
def test_marketing_url_missing(self, mock_get_course_run):
mock_get_course_run.return_value = {}
program_with_type_logo = copy.deepcopy(program)
program_with_type_logo['logo_image'] = program_type['logo_image']
programs_with_type_logo.append(program_with_type_logo)
url = utils.get_run_marketing_url(self.course_key, self.user)
with mock.patch('openedx.core.djangoapps.catalog.utils.get_programs') as patched_get_programs:
with mock.patch('openedx.core.djangoapps.catalog.utils.get_program_types') as patched_get_program_types:
patched_get_programs.return_value = programs
patched_get_program_types.return_value = program_types
self.assertEqual(url, None)
actual = get_programs_with_type_logo()
self.assertEqual(actual, programs_with_type_logo)
openedx/core/djangoapps/catalog/utils.py
"""Helper functions for working with the catalog service."""
from django.conf import settings
from django.contrib.auth.models import User
from django.contrib.auth import get_user_model
from edx_rest_api_client.client import EdxRestApiClient
from opaque_keys.edx.keys import CourseKey
......@@ -9,6 +9,9 @@ from openedx.core.lib.edx_api_utils import get_edx_api_data
from openedx.core.lib.token_utils import JwtBuilder
User = get_user_model() # pylint: disable=invalid-name
def create_catalog_api_client(user, catalog_integration):
"""Returns an API client which can be used to make catalog API requests."""
scopes = ['email', 'profile']
......@@ -18,20 +21,7 @@ def create_catalog_api_client(user, catalog_integration):
return EdxRestApiClient(catalog_integration.internal_api_url, jwt=jwt)
def _get_service_user(user, service_username):
"""
Retrieve and return the Catalog Integration Service User Object
if the passed user is None or anonymous
"""
if not user or user.is_anonymous():
try:
user = User.objects.get(username=service_username)
except User.DoesNotExist:
user = None
return user
def get_programs(user=None, uuid=None, type=None): # pylint: disable=redefined-builtin
def get_programs(uuid=None, type=None): # pylint: disable=redefined-builtin
"""Retrieve marketable programs from the catalog service.
Keyword Arguments:
......@@ -44,8 +34,9 @@ def get_programs(user=None, uuid=None, type=None): # pylint: disable=redefined-
"""
catalog_integration = CatalogIntegration.current()
if catalog_integration.enabled:
user = _get_service_user(user, catalog_integration.service_username)
if not user:
try:
user = User.objects.get(username=catalog_integration.service_username)
except User.DoesNotExist:
return []
api = create_catalog_api_client(user, catalog_integration)
......@@ -75,54 +66,15 @@ def get_programs(user=None, uuid=None, type=None): # pylint: disable=redefined-
return []
def get_program_types(user=None): # pylint: disable=redefined-builtin
"""Retrieve all program types from the catalog service.
Returns:
list of dict, representing program types.
def munge_catalog_program(catalog_program):
"""
catalog_integration = CatalogIntegration.current()
if catalog_integration.enabled:
user = _get_service_user(user, catalog_integration.service_username)
if not user:
return []
api = create_catalog_api_client(user, catalog_integration)
cache_key = '{base}.program_types'.format(base=catalog_integration.CACHE_KEY)
return get_edx_api_data(
catalog_integration,
user,
'program_types',
cache_key=cache_key if catalog_integration.is_cache_enabled else None,
api=api
)
else:
return []
def get_programs_data(user=None):
"""Return the list of Programs after adding the ProgramType Logo Image"""
Make a program from the catalog service look like it came from the programs service.
programs_list = get_programs(user)
program_types = get_program_types(user)
We want to display programs from the catalog service on the LMS. The LMS
originally retrieved all program data from the deprecated programs service.
This temporary utility is here to help incrementally swap out the backend.
program_types_lookup_dict = {program_type["name"]: program_type for program_type in program_types}
for program in programs_list:
program["logo_image"] = program_types_lookup_dict[program["type"]]["logo_image"]
return programs_list
def munge_catalog_program(catalog_program):
"""Make a program from the catalog service look like it came from the programs service.
Catalog-based MicroMasters need to be displayed in the LMS. However, the LMS
currently retrieves all program data from the soon-to-be-retired programs service.
Consuming program data exclusively from the catalog service would have taken more time
than we had prior to the MicroMasters launch. This is a functional middle ground
introduced by ECOM-5460. Cleaning up this debt is tracked by ECOM-4418.
Clean up of this debt is tracked by ECOM-4418.
Arguments:
catalog_program (dict): The catalog service's representation of a program.
......@@ -153,10 +105,11 @@ def munge_catalog_program(catalog_program):
} if course['owners'] else {},
'run_modes': [
{
'course_key': run['key'],
'run_key': CourseKey.from_string(run['key']).run,
'mode_slug': 'verified'
} for run in course['course_runs']
'course_key': course_run['key'],
'run_key': CourseKey.from_string(course_run['key']).run,
'mode_slug': course_run['type'],
'marketing_url': course_run['marketing_url'],
} for course_run in course['course_runs']
],
} for course in catalog_program['courses']
],
......@@ -166,48 +119,48 @@ def munge_catalog_program(catalog_program):
'w435h145': catalog_program['banner_image']['small']['url'],
'w348h116': catalog_program['banner_image']['x-small']['url'],
},
# If a detail URL has been added, we don't want to lose it.
'detail_url': catalog_program.get('detail_url'),
}
def get_course_run(course_key, user):
"""Get a course run's data from the course catalog service.
Arguments:
course_key (CourseKey): Course key object identifying the run whose data we want.
user (User): The user to authenticate as when making requests to the catalog service.
def get_program_types():
"""Retrieve all program types from the catalog service.
Returns:
dict, empty if no data could be retrieved.
list of dict, representing program types.
"""
catalog_integration = CatalogIntegration.current()
if catalog_integration.enabled:
try:
user = User.objects.get(username=catalog_integration.service_username)
except User.DoesNotExist:
return []
api = create_catalog_api_client(user, catalog_integration)
cache_key = '{base}.program_types'.format(base=catalog_integration.CACHE_KEY)
data = get_edx_api_data(
return get_edx_api_data(
catalog_integration,
user,
'course_runs',
resource_id=unicode(course_key),
cache_key=catalog_integration.CACHE_KEY if catalog_integration.is_cache_enabled else None,
api=api,
querystring={'exclude_utm': 1},
'program_types',
cache_key=cache_key if catalog_integration.is_cache_enabled else None,
api=api
)
return data if data else {}
else:
return {}
return []
def get_run_marketing_url(course_key, user):
"""Get a course run's marketing URL from the course catalog service.
def get_programs_with_type_logo():
"""
Join program type logos with programs of corresponding type.
"""
programs_list = get_programs()
program_types = get_program_types()
Arguments:
course_key (CourseKey): Course key object identifying the run whose marketing URL we want.
user (User): The user to authenticate as when making requests to the catalog service.
type_logo_map = {program_type['name']: program_type['logo_image'] for program_type in program_types}
Returns:
string, the marketing URL, or None if no URL is available.
"""
course_run = get_course_run(course_key, user)
return course_run.get('marketing_url')
for program in programs_list:
program['logo_image'] = type_logo_map[program['type']]
return programs_list
openedx/core/djangoapps/credentials/tests/test_utils.py
......@@ -8,7 +8,7 @@ import mock
from nose.plugins.attrib import attr
from provider.constants import CONFIDENTIAL
from openedx.core.djangoapps.catalog.tests import factories as catalog_factories
from openedx.core.djangoapps.catalog.tests.factories import ProgramFactory
from openedx.core.djangoapps.credentials.models import CredentialsApiConfig
from openedx.core.djangoapps.credentials.tests.mixins import CredentialsApiConfigMixin, CredentialsDataMixin
from openedx.core.djangoapps.credentials.utils import (
......@@ -18,8 +18,6 @@ from openedx.core.djangoapps.credentials.utils import (
get_programs_for_credentials
)
from openedx.core.djangoapps.credentials.tests import factories
from openedx.core.djangoapps.programs.tests.mixins import ProgramsApiConfigMixin, ProgramsDataMixin
from openedx.core.djangoapps.programs.models import ProgramsApiConfig
from openedx.core.djangolib.testing.utils import CacheIsolationTestCase, skip_unless_lms
from student.tests.factories import UserFactory
......@@ -37,7 +35,6 @@ class TestCredentialsRetrieval(CredentialsApiConfigMixin, CredentialsDataMixin,
super(TestCredentialsRetrieval, self).setUp()
ClientFactory(name=CredentialsApiConfig.OAUTH2_CLIENT_NAME, client_type=CONFIDENTIAL)
ClientFactory(name=ProgramsApiConfig.OAUTH2_CLIENT_NAME, client_type=CONFIDENTIAL)
self.user = UserFactory()
self.primary_uuid = str(uuid.uuid4())
self.alternate_uuid = str(uuid.uuid4())
......@@ -129,7 +126,7 @@ class TestCredentialsRetrieval(CredentialsApiConfigMixin, CredentialsDataMixin,
}
self.mock_credentials_api(self.user, data=credentials_api_response, reset_url=False)
programs = [
catalog_factories.Program(uuid=primary_uuid), catalog_factories.Program(uuid=alternate_uuid)
ProgramFactory(uuid=primary_uuid), ProgramFactory(uuid=alternate_uuid)
]
with mock.patch("openedx.core.djangoapps.credentials.utils.get_programs_for_credentials") as mock_get_programs:
......@@ -165,7 +162,7 @@ class TestCredentialsRetrieval(CredentialsApiConfigMixin, CredentialsDataMixin,
}
self.mock_credentials_api(self.user, data=credentials_api_response, reset_url=False)
programs = [
catalog_factories.Program(uuid=primary_uuid), catalog_factories.Program(uuid=alternate_uuid)
ProgramFactory(uuid=primary_uuid), ProgramFactory(uuid=alternate_uuid)
]
with mock.patch("openedx.core.djangoapps.credentials.utils.get_programs") as mock_get_programs:
......@@ -199,14 +196,14 @@ class TestCredentialsRetrieval(CredentialsApiConfigMixin, CredentialsDataMixin,
def test_get_program_for_certificates(self):
"""Verify programs data can be retrieved and parsed correctly for certificates."""
programs = [
catalog_factories.Program(uuid=self.primary_uuid),
catalog_factories.Program(uuid=self.alternate_uuid)
ProgramFactory(uuid=self.primary_uuid),
ProgramFactory(uuid=self.alternate_uuid)
]
program_credentials_data = self._expected_program_credentials_data()
with mock.patch("openedx.core.djangoapps.credentials.utils.get_programs") as patched_get_programs:
patched_get_programs.return_value = programs
actual = get_programs_for_credentials(self.user, program_credentials_data)
actual = get_programs_for_credentials(program_credentials_data)
self.assertEqual(len(actual), 2)
self.assertEqual(actual, programs)
......@@ -216,6 +213,6 @@ class TestCredentialsRetrieval(CredentialsApiConfigMixin, CredentialsDataMixin,
program_credentials_data = self._expected_program_credentials_data()
with mock.patch("openedx.core.djangoapps.credentials.utils.get_programs") as patched_get_programs:
patched_get_programs.return_value = []
actual = get_programs_for_credentials(self.user, program_credentials_data)
actual = get_programs_for_credentials(program_credentials_data)
self.assertEqual(actual, [])
openedx/core/djangoapps/credentials/utils.py
......@@ -31,12 +31,11 @@ def get_user_credentials(user):
return credentials
def get_programs_for_credentials(user, programs_credentials):
def get_programs_for_credentials(programs_credentials):
""" Given a user and an iterable of credentials, get corresponding programs
data and return it as a list of dictionaries.
Arguments:
user (User): The user to authenticate as for requesting programs.
programs_credentials (list): List of credentials awarded to the user
for completion of a program.
......@@ -44,7 +43,7 @@ def get_programs_for_credentials(user, programs_credentials):
list, containing programs dictionaries.
"""
certified_programs = []
programs = get_programs(user)
programs = get_programs()
for program in programs:
for credential in programs_credentials:
if program['uuid'] == credential['credential']['program_uuid']:
......@@ -84,7 +83,7 @@ def get_user_program_credentials(user):
log.exception('Invalid credential structure: %r', credential)
if programs_credentials:
programs_credentials_data = get_programs_for_credentials(user, programs_credentials)
programs_credentials_data = get_programs_for_credentials(programs_credentials)
return programs_credentials_data
......
openedx/core/djangoapps/programs/management/commands/backpopulate_program_credentials.py
......@@ -80,10 +80,10 @@ class Command(BaseCommand):
course_runs = set()
for program in programs:
for course in program['courses']:
for run in course['course_runs']:
key = CourseKey.from_string(run['key'])
for course_run in course['course_runs']:
key = CourseKey.from_string(course_run['key'])
course_runs.add(
CourseRun(key, run['type'])
CourseRun(key, course_run['type'])
)
return course_runs
......@@ -97,14 +97,7 @@ class Command(BaseCommand):
status_query = Q(status__in=CertificateStatuses.PASSED_STATUSES)
course_run_query = reduce(
lambda x, y: x | y,
# A course run's type is assumed to indicate which mode must be
# completed in order for the run to count towards program completion.
# This supports the same flexible program construction allowed by the
# old programs service (e.g., completion of an old honor-only run may
# count towards completion of a course in a program). This may change
# in the future to make use of the more rigid set of "applicable seat
# types" associated with each program type in the catalog.
[Q(course_id=run.key, mode=run.type) for run in self.course_runs]
[Q(course_id=course_run.key, mode=course_run.type) for course_run in self.course_runs]
)
query = status_query & course_run_query
......
openedx/core/djangoapps/programs/migrations/0010_auto_20170204_2332.py 0 → 100644
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('programs', '0009_programsapiconfig_marketing_path'),
]
operations = [
migrations.AlterField(
model_name='programsapiconfig',
name='internal_service_url',
field=models.URLField(verbose_name='Internal Service URL', blank=True),
),
migrations.AlterField(
model_name='programsapiconfig',
name='public_service_url',
field=models.URLField(verbose_name='Public Service URL', blank=True),
),
]
openedx/core/djangoapps/programs/models.py
......@@ -19,8 +19,8 @@ class ProgramsApiConfig(ConfigurationModel):
api_version_number = models.IntegerField(verbose_name=_("API Version"))
internal_service_url = models.URLField(verbose_name=_("Internal Service URL"))
public_service_url = models.URLField(verbose_name=_("Public Service URL"))
internal_service_url = models.URLField(verbose_name=_("Internal Service URL"), blank=True)
public_service_url = models.URLField(verbose_name=_("Public Service URL"), blank=True)
marketing_path = models.CharField(
max_length=255,
......
openedx/core/djangoapps/programs/tasks/v1/tasks.py
......@@ -65,7 +65,7 @@ def get_completed_programs(student):
list of program UUIDs
"""
meter = ProgramProgressMeter(student, use_catalog=True)
meter = ProgramProgressMeter(student)
return meter.completed_programs
......
openedx/core/djangoapps/programs/tasks/v1/tests/test_tasks.py
......@@ -6,24 +6,20 @@ import json
from celery.exceptions import MaxRetriesExceededError
import ddt
from django.conf import settings
from django.core.cache import cache
from django.test import override_settings, TestCase
from edx_rest_api_client.client import EdxRestApiClient
from edx_oauth2_provider.tests.factories import ClientFactory
import httpretty
import mock
from provider.constants import CONFIDENTIAL
from lms.djangoapps.certificates.api import MODES
from openedx.core.djangoapps.catalog.tests import factories, mixins
from openedx.core.djangoapps.catalog.utils import munge_catalog_program
from openedx.core.djangoapps.catalog.tests.mixins import CatalogIntegrationMixin
from openedx.core.djangoapps.credentials.tests.mixins import CredentialsApiConfigMixin
from openedx.core.djangoapps.programs.tasks.v1 import tasks
from openedx.core.djangolib.testing.utils import CacheIsolationTestCase, skip_unless_lms
from openedx.core.djangolib.testing.utils import skip_unless_lms
from student.tests.factories import UserFactory
TASKS_MODULE = 'openedx.core.djangoapps.programs.tasks.v1.tasks'
UTILS_MODULE = 'openedx.core.djangoapps.programs.utils'
@skip_unless_lms
......@@ -49,57 +45,6 @@ class GetApiClientTestCase(CredentialsApiConfigMixin, TestCase):
self.assertEqual(api_client._store['session'].auth.token, 'test-token') # pylint: disable=protected-access
@httpretty.activate
@skip_unless_lms
class GetCompletedProgramsTestCase(mixins.CatalogIntegrationMixin, CacheIsolationTestCase):
"""
Test the get_completed_programs function
"""
ENABLED_CACHES = ['default']
def setUp(self):
super(GetCompletedProgramsTestCase, self).setUp()
self.user = UserFactory()
self.catalog_integration = self.create_catalog_integration(cache_ttl=1)
def _mock_programs_api(self, data):
"""Helper for mocking out Programs API URLs."""
self.assertTrue(httpretty.is_enabled(), msg='httpretty must be enabled to mock API calls.')
url = self.catalog_integration.internal_api_url.strip('/') + '/programs/'
body = json.dumps({'results': data})
httpretty.register_uri(httpretty.GET, url, body=body, content_type='application/json')
def _assert_num_requests(self, count):
"""DRY helper for verifying request counts."""
self.assertEqual(len(httpretty.httpretty.latest_requests), count)
@mock.patch(UTILS_MODULE + '.get_completed_courses')
def test_get_completed_programs(self, mock_get_completed_courses):
"""
Verify that completed programs are found, using the cache when possible.
"""
data = [
factories.Program(),
]
self._mock_programs_api(data)
munged_program = munge_catalog_program(data[0])
course_codes = munged_program['course_codes']
mock_get_completed_courses.return_value = [
{'course_id': run_mode['course_key'], 'mode': run_mode['mode_slug']}
for run_mode in course_codes[0]['run_modes']
]
for _ in range(2):
result = tasks.get_completed_programs(self.user)
self.assertEqual(result[0], munged_program['id'])
# Verify that only one request to the catalog was made (i.e., the cache was hit).
self._assert_num_requests(1)
@skip_unless_lms
class GetAwardedCertificateProgramsTestCase(TestCase):
"""
......@@ -175,7 +120,7 @@ class AwardProgramCertificateTestCase(TestCase):
@mock.patch(TASKS_MODULE + '.get_certified_programs')
@mock.patch(TASKS_MODULE + '.get_completed_programs')
@override_settings(CREDENTIALS_SERVICE_USERNAME='test-service-username')
class AwardProgramCertificatesTestCase(mixins.CatalogIntegrationMixin, CredentialsApiConfigMixin, TestCase):
class AwardProgramCertificatesTestCase(CatalogIntegrationMixin, CredentialsApiConfigMixin, TestCase):
"""
Tests for the 'award_program_certificates' celery task.
"""
......
openedx/core/djangoapps/programs/tests/factories.py
"""Factories for generating fake program-related data."""
# pylint: disable=missing-docstring, invalid-name
import factory
from factory.fuzzy import FuzzyText
from faker import Faker
class Program(factory.Factory):
"""
Factory for stubbing program resources from the Programs API (v1).
"""
class Meta(object):
model = dict
id = factory.Sequence(lambda n: n) # pylint: disable=invalid-name
name = FuzzyText(prefix='Program ')
subtitle = FuzzyText(prefix='Subtitle ')
category = 'FooBar'
status = 'active'
marketing_slug = FuzzyText(prefix='slug_')
organizations = []
course_codes = []
banner_image_urls = {}
class Organization(factory.Factory):
"""
Factory for stubbing nested organization resources from the Programs API (v1).
"""
class Meta(object):
model = dict
key = FuzzyText(prefix='org_')
display_name = FuzzyText(prefix='Display Name ')
class CourseCode(factory.Factory):
"""
Factory for stubbing nested course code resources from the Programs API (v1).
"""
class Meta(object):
model = dict
display_name = FuzzyText(prefix='Display Name ')
run_modes = []
class RunMode(factory.Factory):
"""
Factory for stubbing nested run mode resources from the Programs API (v1).
"""
class Meta(object):
model = dict
course_key = FuzzyText(prefix='org/', suffix='/run')
mode_slug = 'verified'
fake = Faker()
class Progress(factory.Factory):
"""
Factory for stubbing program progress dicts.
"""
class ProgressFactory(factory.Factory):
class Meta(object):
model = dict
id = factory.Sequence(lambda n: n) # pylint: disable=invalid-name
uuid = factory.Faker('uuid4')
completed = []
in_progress = []
not_started = []
openedx/core/djangoapps/programs/tests/mixins.py
"""Mixins for use during testing."""
import json
import httpretty
from openedx.core.djangoapps.programs.models import ProgramsApiConfig
from openedx.core.djangoapps.programs.tests import factories
class ProgramsApiConfigMixin(object):
......@@ -29,87 +24,3 @@ class ProgramsApiConfigMixin(object):
ProgramsApiConfig(**fields).save()
return ProgramsApiConfig.current()
class ProgramsDataMixin(object):
"""Mixin mocking Programs API URLs and providing fake data for testing.
NOTE: This mixin is DEPRECATED. Tests should create and manage their own data.
"""
PROGRAM_NAMES = [
'Test Program A',
'Test Program B',
'Test Program C',
]
COURSE_KEYS = [
'organization-a/course-a/fall',
'organization-a/course-a/winter',
'organization-a/course-b/fall',
'organization-a/course-b/winter',
'organization-b/course-c/fall',
'organization-b/course-c/winter',
'organization-b/course-d/fall',
'organization-b/course-d/winter',
]
PROGRAMS_API_RESPONSE = {
'results': [
factories.Program(
id=1,
name=PROGRAM_NAMES[0],
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[
factories.RunMode(course_key=COURSE_KEYS[0]),
factories.RunMode(course_key=COURSE_KEYS[1]),
]),
factories.CourseCode(run_modes=[
factories.RunMode(course_key=COURSE_KEYS[2]),
factories.RunMode(course_key=COURSE_KEYS[3]),
]),
]
),
factories.Program(
id=2,
name=PROGRAM_NAMES[1],
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[
factories.RunMode(course_key=COURSE_KEYS[4]),
factories.RunMode(course_key=COURSE_KEYS[5]),
]),
factories.CourseCode(run_modes=[
factories.RunMode(course_key=COURSE_KEYS[6]),
factories.RunMode(course_key=COURSE_KEYS[7]),
]),
]
),
factories.Program(
id=3,
name=PROGRAM_NAMES[2],
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[
factories.RunMode(course_key=COURSE_KEYS[7]),
]),
]
),
]
}
def mock_programs_api(self, data=None, program_id='', status_code=200):
"""Utility for mocking out Programs API URLs."""
self.assertTrue(httpretty.is_enabled(), msg='httpretty must be enabled to mock Programs API calls.')
url = ProgramsApiConfig.current().internal_api_url.strip('/') + '/programs/'
if program_id:
url += '{}/'.format(str(program_id))
if data is None:
data = self.PROGRAMS_API_RESPONSE
body = json.dumps(data)
httpretty.reset()
httpretty.register_uri(httpretty.GET, url, body=body, content_type='application/json', status=status_code)
openedx/core/djangoapps/programs/tests/test_backpopulate_program_credentials.py
......@@ -7,7 +7,12 @@ import mock
from certificates.models import CertificateStatuses # pylint: disable=import-error
from lms.djangoapps.certificates.api import MODES
from lms.djangoapps.certificates.tests.factories import GeneratedCertificateFactory
from openedx.core.djangoapps.catalog.tests import factories
from openedx.core.djangoapps.catalog.tests.factories import (
generate_course_run_key,
ProgramFactory,
CourseFactory,
CourseRunFactory,
)
from openedx.core.djangoapps.catalog.tests.mixins import CatalogIntegrationMixin
from openedx.core.djangoapps.credentials.tests.mixins import CredentialsApiConfigMixin
from openedx.core.djangolib.testing.utils import skip_unless_lms
......@@ -23,7 +28,7 @@ COMMAND_MODULE = 'openedx.core.djangoapps.programs.management.commands.backpopul
@skip_unless_lms
class BackpopulateProgramCredentialsTests(CatalogIntegrationMixin, CredentialsApiConfigMixin, TestCase):
"""Tests for the backpopulate_program_credentials management command."""
course_run_key, alternate_course_run_key = (factories.generate_course_run_key() for __ in range(2))
course_run_key, alternate_course_run_key = (generate_course_run_key() for __ in range(2))
def setUp(self):
super(BackpopulateProgramCredentialsTests, self).setUp()
......@@ -36,8 +41,8 @@ class BackpopulateProgramCredentialsTests(CatalogIntegrationMixin, CredentialsAp
# skewing mock call counts.
self.create_credentials_config(enable_learner_issuance=False)
self.catalog_integration = self.create_catalog_integration()
self.service_user = UserFactory(username=self.catalog_integration.service_username)
catalog_integration = self.create_catalog_integration()
UserFactory(username=catalog_integration.service_username)
@ddt.data(True, False)
def test_handle(self, commit, mock_task, mock_get_programs):
......@@ -45,10 +50,10 @@ class BackpopulateProgramCredentialsTests(CatalogIntegrationMixin, CredentialsAp
Verify that relevant tasks are only enqueued when the commit option is passed.
"""
data = [
factories.Program(
ProgramFactory(
courses=[
factories.Course(course_runs=[
factories.CourseRun(key=self.course_run_key),
CourseFactory(course_runs=[
CourseRunFactory(key=self.course_run_key),
]),
]
),
......@@ -78,39 +83,39 @@ class BackpopulateProgramCredentialsTests(CatalogIntegrationMixin, CredentialsAp
@ddt.data(
[
factories.Program(
ProgramFactory(
courses=[
factories.Course(course_runs=[
factories.CourseRun(key=course_run_key),
CourseFactory(course_runs=[
CourseRunFactory(key=course_run_key),
]),
]
),
factories.Program(
ProgramFactory(
courses=[
factories.Course(course_runs=[
factories.CourseRun(key=alternate_course_run_key),
CourseFactory(course_runs=[
CourseRunFactory(key=alternate_course_run_key),
]),
]
),
],
[
factories.Program(
ProgramFactory(
courses=[
factories.Course(course_runs=[
factories.CourseRun(key=course_run_key),
CourseFactory(course_runs=[
CourseRunFactory(key=course_run_key),
]),
factories.Course(course_runs=[
factories.CourseRun(key=alternate_course_run_key),
CourseFactory(course_runs=[
CourseRunFactory(key=alternate_course_run_key),
]),
]
),
],
[
factories.Program(
ProgramFactory(
courses=[
factories.Course(course_runs=[
factories.CourseRun(key=course_run_key),
factories.CourseRun(key=alternate_course_run_key),
CourseFactory(course_runs=[
CourseRunFactory(key=course_run_key),
CourseRunFactory(key=alternate_course_run_key),
]),
]
),
......@@ -148,11 +153,11 @@ class BackpopulateProgramCredentialsTests(CatalogIntegrationMixin, CredentialsAp
course run certificates.
"""
data = [
factories.Program(
ProgramFactory(
courses=[
factories.Course(course_runs=[
factories.CourseRun(key=self.course_run_key),
factories.CourseRun(key=self.alternate_course_run_key),
CourseFactory(course_runs=[
CourseRunFactory(key=self.course_run_key),
CourseRunFactory(key=self.alternate_course_run_key),
]),
]
),
......@@ -183,10 +188,10 @@ class BackpopulateProgramCredentialsTests(CatalogIntegrationMixin, CredentialsAp
qualifying course run certificates.
"""
data = [
factories.Program(
ProgramFactory(
courses=[
factories.Course(course_runs=[
factories.CourseRun(key=self.course_run_key, type='honor'),
CourseFactory(course_runs=[
CourseRunFactory(key=self.course_run_key, type='honor'),
]),
]
),
......@@ -216,10 +221,10 @@ class BackpopulateProgramCredentialsTests(CatalogIntegrationMixin, CredentialsAp
Verify that only course run certificates with a passing status are selected.
"""
data = [
factories.Program(
ProgramFactory(
courses=[
factories.Course(course_runs=[
factories.CourseRun(key=self.course_run_key),
CourseFactory(course_runs=[
CourseRunFactory(key=self.course_run_key),
]),
]
),
......@@ -261,10 +266,10 @@ class BackpopulateProgramCredentialsTests(CatalogIntegrationMixin, CredentialsAp
mock_task.side_effect = side_effect
data = [
factories.Program(
ProgramFactory(
courses=[
factories.Course(course_runs=[
factories.CourseRun(key=self.course_run_key),
CourseFactory(course_runs=[
CourseRunFactory(key=self.course_run_key),
]),
]
),
......
openedx/core/djangoapps/programs/tests/test_utils.py
"""Tests covering Programs utilities."""
import copy
import datetime
import json
import uuid
......@@ -9,630 +8,413 @@ from django.core.cache import cache
from django.core.urlresolvers import reverse
from django.test import TestCase
from django.test.utils import override_settings
from django.utils.text import slugify
import httpretty
import mock
from nose.plugins.attrib import attr
from opaque_keys.edx.keys import CourseKey
from edx_oauth2_provider.tests.factories import ClientFactory
from provider.constants import CONFIDENTIAL
from pytz import utc
from lms.djangoapps.certificates.api import MODES
from lms.djangoapps.commerce.tests.test_utils import update_commerce_config
from openedx.core.djangoapps.catalog.utils import munge_catalog_program
from openedx.core.djangoapps.catalog.tests.factories import (
generate_course_run_key,
ProgramFactory,
CourseFactory,
CourseRunFactory,
OrganizationFactory,
)
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from openedx.core.djangoapps.credentials.tests.mixins import CredentialsApiConfigMixin, CredentialsDataMixin
from openedx.core.djangoapps.programs import utils
from openedx.core.djangoapps.programs.models import ProgramsApiConfig
from openedx.core.djangoapps.programs.tests import factories
from openedx.core.djangoapps.programs.tests.mixins import ProgramsApiConfigMixin, ProgramsDataMixin
from openedx.core.djangoapps.programs.tests.factories import ProgressFactory
from openedx.core.djangoapps.programs.utils import (
DEFAULT_ENROLLMENT_START_DATE, ProgramProgressMeter, ProgramDataExtender
)
from openedx.core.djangolib.testing.utils import CacheIsolationTestCase, skip_unless_lms
from student.tests.factories import UserFactory, CourseEnrollmentFactory
from util.date_utils import strftime_localized
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
from xmodule.modulestore.tests.factories import CourseFactory as ModuleStoreCourseFactory
UTILS_MODULE = 'openedx.core.djangoapps.programs.utils'
CERTIFICATES_API_MODULE = 'lms.djangoapps.certificates.api'
ECOMMERCE_URL_ROOT = 'https://example-ecommerce.com'
MARKETING_URL = 'https://www.example.com/marketing/path'
@ddt.ddt
@attr(shard=2)
@httpretty.activate
@skip_unless_lms
class TestProgramRetrieval(ProgramsApiConfigMixin, ProgramsDataMixin, CredentialsDataMixin,
CredentialsApiConfigMixin, CacheIsolationTestCase):
"""Tests covering the retrieval of programs from the Programs service."""
ENABLED_CACHES = ['default']
def setUp(self):
super(TestProgramRetrieval, self).setUp()
ClientFactory(name=ProgramsApiConfig.OAUTH2_CLIENT_NAME, client_type=CONFIDENTIAL)
self.user = UserFactory()
cache.clear()
def test_get_programs(self):
"""Verify programs data can be retrieved."""
self.create_programs_config()
self.mock_programs_api()
actual = utils.get_programs(self.user)
self.assertEqual(
actual,
self.PROGRAMS_API_RESPONSE['results']
)
# Verify the API was actually hit (not the cache).
self.assertEqual(len(httpretty.httpretty.latest_requests), 1)
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.
utils.get_programs(self.user)
# Hit the cache.
utils.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):
utils.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 = utils.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 = utils.get_programs(self.user)
self.assertEqual(actual, [])
self.assertTrue(mock_init.called)
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 = utils.get_programs(self.user)
self.assertEqual(actual, [])
@skip_unless_lms
class GetProgramsByRunTests(TestCase):
"""Tests verifying that programs are inverted correctly."""
maxDiff = None
@classmethod
def setUpClass(cls):
super(GetProgramsByRunTests, cls).setUpClass()
cls.user = UserFactory()
course_keys = [
CourseKey.from_string('some/course/run'),
CourseKey.from_string('some/other/run'),
]
cls.enrollments = [CourseEnrollmentFactory(user=cls.user, course_id=c) for c in course_keys]
cls.course_ids = [unicode(c) for c in course_keys]
organization = factories.Organization()
joint_programs = sorted([
factories.Program(
organizations=[organization],
course_codes=[
factories.CourseCode(run_modes=[
factories.RunMode(course_key=cls.course_ids[0]),
]),
]
) for __ in range(2)
], key=lambda p: p['name'])
cls.programs = joint_programs + [
factories.Program(
organizations=[organization],
course_codes=[
factories.CourseCode(run_modes=[
factories.RunMode(course_key=cls.course_ids[1]),
]),
]
),
factories.Program(
organizations=[organization],
course_codes=[
factories.CourseCode(run_modes=[
factories.RunMode(course_key='yet/another/run'),
]),
]
),
]
def test_get_programs_by_run(self):
"""Verify that programs are organized by run ID."""
programs_by_run, course_ids = utils.get_programs_by_run(self.programs, self.enrollments)
self.assertEqual(programs_by_run[self.course_ids[0]], self.programs[:2])
self.assertEqual(programs_by_run[self.course_ids[1]], self.programs[2:3])
self.assertEqual(course_ids, self.course_ids)
def test_no_programs(self):
"""Verify that the utility can cope with missing programs data."""
programs_by_run, course_ids = utils.get_programs_by_run([], self.enrollments)
self.assertEqual(programs_by_run, {})
self.assertEqual(course_ids, self.course_ids)
def test_no_enrollments(self):
"""Verify that the utility can cope with missing enrollment data."""
programs_by_run, course_ids = utils.get_programs_by_run(self.programs, [])
self.assertEqual(programs_by_run, {})
self.assertEqual(course_ids, [])
@skip_unless_lms
class GetCompletedCoursesTestCase(TestCase):
"""
Test the get_completed_courses function
"""
def make_cert_result(self, **kwargs):
"""
Helper to create dummy results from the certificates API
"""
result = {
'username': 'dummy-username',
'course_key': 'dummy-course',
'type': 'dummy-type',
'status': 'dummy-status',
'download_url': 'http://www.example.com/cert.pdf',
'grade': '0.98',
'created': '2015-07-31T00:00:00Z',
'modified': '2015-07-31T00:00:00Z',
}
result.update(**kwargs)
return result
@mock.patch(UTILS_MODULE + '.certificate_api.get_certificates_for_user')
def test_get_completed_courses(self, mock_get_certs_for_user):
"""
Ensure the function correctly calls to and handles results from the
certificates API
"""
student = UserFactory(username='test-username')
mock_get_certs_for_user.return_value = [
self.make_cert_result(status='downloadable', type='verified', course_key='downloadable-course'),
self.make_cert_result(status='generating', type='professional', course_key='generating-course'),
self.make_cert_result(status='unknown', type='honor', course_key='unknown-course'),
]
result = utils.get_completed_courses(student)
self.assertEqual(mock_get_certs_for_user.call_args[0], (student.username, ))
self.assertEqual(result, [
{'course_id': 'downloadable-course', 'mode': 'verified'},
{'course_id': 'generating-course', 'mode': 'professional'},
])
@attr(shard=2)
@httpretty.activate
@skip_unless_lms
class TestProgramProgressMeter(ProgramsApiConfigMixin, TestCase):
@mock.patch(UTILS_MODULE + '.get_programs')
class TestProgramProgressMeter(TestCase):
"""Tests of the program progress utility class."""
def setUp(self):
super(TestProgramProgressMeter, self).setUp()
self.user = UserFactory()
self.create_programs_config()
ClientFactory(name=ProgramsApiConfig.OAUTH2_CLIENT_NAME, client_type=CONFIDENTIAL)
def _mock_programs_api(self, data):
"""Helper for mocking out Programs API URLs."""
self.assertTrue(httpretty.is_enabled(), msg='httpretty must be enabled to mock Programs API calls.')
url = ProgramsApiConfig.current().internal_api_url.strip('/') + '/programs/'
body = json.dumps({'results': data})
httpretty.register_uri(httpretty.GET, url, body=body, content_type='application/json')
def _create_enrollments(self, *course_ids):
"""Variadic helper used to create course enrollments."""
for course_id in course_ids:
CourseEnrollmentFactory(user=self.user, course_id=course_id)
def _create_enrollments(self, *course_run_ids):
"""Variadic helper used to create course run enrollments."""
for course_run_id in course_run_ids:
CourseEnrollmentFactory(user=self.user, course_id=course_run_id)
def _assert_progress(self, meter, *progresses):
"""Variadic helper used to verify progress calculations."""
self.assertEqual(meter.progress, list(progresses))
def _extract_names(self, program, *course_codes):
"""Construct a list containing the display names of the indicated course codes."""
return [program['course_codes'][cc]['display_name'] for cc in course_codes]
def _extract_titles(self, program, *indices):
"""Construct a list containing the titles of the indicated courses."""
return [program['courses'][index]['title'] for index in indices]
def _attach_detail_url(self, programs):
"""Add expected detail URLs to a list of program dicts."""
for program in programs:
base = reverse('program_details_view', kwargs={'program_id': program['id']}).rstrip('/')
slug = slugify(program['name'])
program['detail_url'] = reverse('program_details_view', kwargs={'program_uuid': program['uuid']})
program['detail_url'] = '{base}/{slug}'.format(base=base, slug=slug)
def test_no_enrollments(self):
def test_no_enrollments(self, mock_get_programs):
"""Verify behavior when programs exist, but no relevant enrollments do."""
data = [
factories.Program(
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[factories.RunMode()]),
]
),
]
self._mock_programs_api(data)
data = [ProgramFactory()]
mock_get_programs.return_value = data
meter = utils.ProgramProgressMeter(self.user)
meter = ProgramProgressMeter(self.user)
self.assertEqual(meter.engaged_programs(), [])
self.assertEqual(meter.engaged_programs, [])
self._assert_progress(meter)
self.assertEqual(meter.completed_programs, [])
def test_no_programs(self):
def test_no_programs(self, mock_get_programs):
"""Verify behavior when enrollments exist, but no matching programs do."""
self._mock_programs_api([])
mock_get_programs.return_value = []
self._create_enrollments('org/course/run')
meter = utils.ProgramProgressMeter(self.user)
course_run_id = generate_course_run_key()
self._create_enrollments(course_run_id)
meter = ProgramProgressMeter(self.user)
self.assertEqual(meter.engaged_programs(), [])
self.assertEqual(meter.engaged_programs, [])
self._assert_progress(meter)
self.assertEqual(meter.completed_programs, [])
def test_single_program_engagement(self):
def test_single_program_engagement(self, mock_get_programs):
"""
Verify that correct program is returned when the user has a single enrollment
appearing in one program.
Verify that correct program is returned when the user is enrolled in a
course run appearing in one program.
"""
course_id = 'org/course/run'
course_run_key = generate_course_run_key()
data = [
factories.Program(
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[
factories.RunMode(course_key=course_id),
ProgramFactory(
courses=[
CourseFactory(course_runs=[
CourseRunFactory(key=course_run_key),
]),
]
),
factories.Program(
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[factories.RunMode()]),
]
),
ProgramFactory(),
]
self._mock_programs_api(data)
mock_get_programs.return_value = data
self._create_enrollments(course_id)
meter = utils.ProgramProgressMeter(self.user)
self._create_enrollments(course_run_key)
meter = ProgramProgressMeter(self.user)
self._attach_detail_url(data)
program = data[0]
self.assertEqual(meter.engaged_programs(), [program])
self.assertEqual(meter.engaged_programs, [program])
self._assert_progress(
meter,
factories.Progress(
id=program['id'],
in_progress=self._extract_names(program, 0)
ProgressFactory(
uuid=program['uuid'],
in_progress=self._extract_titles(program, 0)
)
)
self.assertEqual(meter.completed_programs, [])
def test_mutiple_program_engagement(self):
def test_mutiple_program_engagement(self, mock_get_programs):
"""
Verify that correct programs are returned in the correct order when the user
has multiple enrollments.
Verify that correct programs are returned in the correct order when the
user is enrolled in course runs appearing in programs.
"""
first_course_id, second_course_id = 'org/first-course/run', 'org/second-course/run'
newer_course_run_key, older_course_run_key = (generate_course_run_key() for __ in range(2))
data = [
factories.Program(
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[
factories.RunMode(course_key=first_course_id),
ProgramFactory(
courses=[
CourseFactory(course_runs=[
CourseRunFactory(key=newer_course_run_key),
]),
]
),
factories.Program(
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[
factories.RunMode(course_key=second_course_id),
ProgramFactory(
courses=[
CourseFactory(course_runs=[
CourseRunFactory(key=older_course_run_key),
]),
]
),
factories.Program(
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[factories.RunMode()]),
ProgramFactory(),
]
),
]
self._mock_programs_api(data)
mock_get_programs.return_value = data
self._create_enrollments(second_course_id, first_course_id)
meter = utils.ProgramProgressMeter(self.user)
# The creation time of the enrollments matters to the test. We want
# the first_course_run_key to represent the newest enrollment.
self._create_enrollments(older_course_run_key, newer_course_run_key)
meter = ProgramProgressMeter(self.user)
self._attach_detail_url(data)
programs = data[:2]
self.assertEqual(meter.engaged_programs(), programs)
self.assertEqual(meter.engaged_programs, programs)
self._assert_progress(
meter,
factories.Progress(id=programs[0]['id'], in_progress=self._extract_names(programs[0], 0)),
factories.Progress(id=programs[1]['id'], in_progress=self._extract_names(programs[1], 0))
*(
ProgressFactory(uuid=program['uuid'], in_progress=self._extract_titles(program, 0))
for program in programs
)
)
self.assertEqual(meter.completed_programs, [])
def test_shared_enrollment_engagement(self):
def test_shared_enrollment_engagement(self, mock_get_programs):
"""
Verify that correct programs are returned when the user has a single enrollment
appearing in multiple programs.
Verify that correct programs are returned when the user is enrolled in a
single course run appearing in multiple programs.
"""
shared_course_id, solo_course_id = 'org/shared-course/run', 'org/solo-course/run'
joint_programs = sorted([
factories.Program(
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[
factories.RunMode(course_key=shared_course_id),
shared_course_run_key, solo_course_run_key = (generate_course_run_key() for __ in range(2))
batch = [
ProgramFactory(
courses=[
CourseFactory(course_runs=[
CourseRunFactory(key=shared_course_run_key),
]),
]
) for __ in range(2)
], key=lambda p: p['name'])
)
for __ in range(2)
]
joint_programs = sorted(batch, key=lambda program: program['title'])
data = joint_programs + [
factories.Program(
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[
factories.RunMode(course_key=solo_course_id),
ProgramFactory(
courses=[
CourseFactory(course_runs=[
CourseRunFactory(key=solo_course_run_key),
]),
]
),
factories.Program(
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[factories.RunMode()]),
]
),
ProgramFactory(),
]
self._mock_programs_api(data)
mock_get_programs.return_value = data
# Enrollment for the shared course ID created last (most recently).
self._create_enrollments(solo_course_id, shared_course_id)
meter = utils.ProgramProgressMeter(self.user)
# Enrollment for the shared course run created last (most recently).
self._create_enrollments(solo_course_run_key, shared_course_run_key)
meter = ProgramProgressMeter(self.user)
self._attach_detail_url(data)
programs = data[:3]
self.assertEqual(meter.engaged_programs(), programs)
self.assertEqual(meter.engaged_programs, programs)
self._assert_progress(
meter,
factories.Progress(id=programs[0]['id'], in_progress=self._extract_names(programs[0], 0)),
factories.Progress(id=programs[1]['id'], in_progress=self._extract_names(programs[1], 0)),
factories.Progress(id=programs[2]['id'], in_progress=self._extract_names(programs[2], 0))
*(
ProgressFactory(uuid=program['uuid'], in_progress=self._extract_titles(program, 0))
for program in programs
)
)
self.assertEqual(meter.completed_programs, [])
@mock.patch(UTILS_MODULE + '.get_completed_courses')
def test_simulate_progress(self, mock_get_completed_courses):
@mock.patch(UTILS_MODULE + '.ProgramProgressMeter.completed_course_runs', new_callable=mock.PropertyMock)
def test_simulate_progress(self, mock_completed_course_runs, mock_get_programs):
"""Simulate the entirety of a user's progress through a program."""
first_course_id, second_course_id = 'org/first-course/run', 'org/second-course/run'
first_course_run_key, second_course_run_key = (generate_course_run_key() for __ in range(2))
data = [
factories.Program(
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[
factories.RunMode(course_key=first_course_id),
ProgramFactory(
courses=[
CourseFactory(course_runs=[
CourseRunFactory(key=first_course_run_key),
]),
factories.CourseCode(run_modes=[
factories.RunMode(course_key=second_course_id),
CourseFactory(course_runs=[
CourseRunFactory(key=second_course_run_key),
]),
]
),
factories.Program(
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[factories.RunMode()]),
]
),
ProgramFactory(),
]
self._mock_programs_api(data)
mock_get_programs.return_value = data
# No enrollments, no program engaged.
meter = utils.ProgramProgressMeter(self.user)
# No enrollments, no programs in progress.
meter = ProgramProgressMeter(self.user)
self._assert_progress(meter)
self.assertEqual(meter.completed_programs, [])
# One enrollment, program engaged.
self._create_enrollments(first_course_id)
meter = utils.ProgramProgressMeter(self.user)
program, program_id = data[0], data[0]['id']
# One enrollment, one program in progress.
self._create_enrollments(first_course_run_key)
meter = ProgramProgressMeter(self.user)
program, program_uuid = data[0], data[0]['uuid']
self._assert_progress(
meter,
factories.Progress(
id=program_id,
in_progress=self._extract_names(program, 0),
not_started=self._extract_names(program, 1)
ProgressFactory(
uuid=program_uuid,
in_progress=self._extract_titles(program, 0),
not_started=self._extract_titles(program, 1)
)
)
self.assertEqual(meter.completed_programs, [])
# Two enrollments, program in progress.
self._create_enrollments(second_course_id)
meter = utils.ProgramProgressMeter(self.user)
# Two enrollments, all courses in progress.
self._create_enrollments(second_course_run_key)
meter = ProgramProgressMeter(self.user)
self._assert_progress(
meter,
factories.Progress(
id=program_id,
in_progress=self._extract_names(program, 0, 1)
ProgressFactory(
uuid=program_uuid,
in_progress=self._extract_titles(program, 0, 1)
)
)
self.assertEqual(meter.completed_programs, [])
# One valid certificate earned, one course code complete.
mock_get_completed_courses.return_value = [
{'course_id': first_course_id, 'mode': MODES.verified},
# One valid certificate earned, one course complete.
mock_completed_course_runs.return_value = [
{'course_run_id': first_course_run_key, 'type': MODES.verified},
]
meter = utils.ProgramProgressMeter(self.user)
meter = ProgramProgressMeter(self.user)
self._assert_progress(
meter,
factories.Progress(
id=program_id,
completed=self._extract_names(program, 0),
in_progress=self._extract_names(program, 1)
ProgressFactory(
uuid=program_uuid,
completed=self._extract_titles(program, 0),
in_progress=self._extract_titles(program, 1)
)
)
self.assertEqual(meter.completed_programs, [])
# Invalid certificate earned, still one course code to complete.
mock_get_completed_courses.return_value = [
{'course_id': first_course_id, 'mode': MODES.verified},
{'course_id': second_course_id, 'mode': MODES.honor},
# Invalid certificate earned, still one course to complete.
mock_completed_course_runs.return_value = [
{'course_run_id': first_course_run_key, 'type': MODES.verified},
{'course_run_id': second_course_run_key, 'type': MODES.honor},
]
meter = utils.ProgramProgressMeter(self.user)
meter = ProgramProgressMeter(self.user)
self._assert_progress(
meter,
factories.Progress(
id=program_id,
completed=self._extract_names(program, 0),
in_progress=self._extract_names(program, 1)
ProgressFactory(
uuid=program_uuid,
completed=self._extract_titles(program, 0),
in_progress=self._extract_titles(program, 1)
)
)
self.assertEqual(meter.completed_programs, [])
# Second valid certificate obtained, all course codes complete.
mock_get_completed_courses.return_value = [
{'course_id': first_course_id, 'mode': MODES.verified},
{'course_id': second_course_id, 'mode': MODES.verified},
# Second valid certificate obtained, all courses complete.
mock_completed_course_runs.return_value = [
{'course_run_id': first_course_run_key, 'type': MODES.verified},
{'course_run_id': second_course_run_key, 'type': MODES.verified},
]
meter = utils.ProgramProgressMeter(self.user)
meter = ProgramProgressMeter(self.user)
self._assert_progress(
meter,
factories.Progress(
id=program_id,
completed=self._extract_names(program, 0, 1)
ProgressFactory(
uuid=program_uuid,
completed=self._extract_titles(program, 0, 1)
)
)
self.assertEqual(meter.completed_programs, [program_id])
self.assertEqual(meter.completed_programs, [program_uuid])
@mock.patch(UTILS_MODULE + '.get_completed_courses')
def test_nonstandard_run_mode_completion(self, mock_get_completed_courses):
@mock.patch(UTILS_MODULE + '.ProgramProgressMeter.completed_course_runs', new_callable=mock.PropertyMock)
def test_nonverified_course_run_completion(self, mock_completed_course_runs, mock_get_programs):
"""
A valid run mode isn't necessarily verified. Verify that a program can
Course runs aren't necessarily of type verified. Verify that a program can
still be completed when this is the case.
"""
course_id = 'org/course/run'
course_run_key = generate_course_run_key()
data = [
factories.Program(
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[
factories.RunMode(
course_key=course_id,
mode_slug=MODES.honor
),
factories.RunMode(),
ProgramFactory(
courses=[
CourseFactory(course_runs=[
CourseRunFactory(key=course_run_key, type='honor'),
CourseRunFactory(),
]),
]
),
factories.Program(
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[factories.RunMode()]),
ProgramFactory(),
]
),
]
self._mock_programs_api(data)
mock_get_programs.return_value = data
self._create_enrollments(course_id)
mock_get_completed_courses.return_value = [
{'course_id': course_id, 'mode': MODES.honor},
self._create_enrollments(course_run_key)
mock_completed_course_runs.return_value = [
{'course_run_id': course_run_key, 'type': MODES.honor},
]
meter = utils.ProgramProgressMeter(self.user)
meter = ProgramProgressMeter(self.user)
program, program_id = data[0], data[0]['id']
program, program_uuid = data[0], data[0]['uuid']
self._assert_progress(
meter,
factories.Progress(id=program_id, completed=self._extract_names(program, 0))
ProgressFactory(uuid=program_uuid, completed=self._extract_titles(program, 0))
)
self.assertEqual(meter.completed_programs, [program_id])
self.assertEqual(meter.completed_programs, [program_uuid])
@mock.patch(UTILS_MODULE + '.get_completed_courses')
def test_completed_programs(self, mock_get_completed_courses):
@mock.patch(UTILS_MODULE + '.ProgramProgressMeter.completed_course_runs', new_callable=mock.PropertyMock)
def test_completed_programs(self, mock_completed_course_runs, mock_get_programs):
"""Verify that completed programs are correctly identified."""
program_count, course_code_count, run_mode_count = 3, 2, 2
data = [
factories.Program(
organizations=[factories.Organization()],
course_codes=[
factories.CourseCode(run_modes=[factories.RunMode() for _ in range(run_mode_count)])
for _ in range(course_code_count)
]
)
for _ in range(program_count)
]
self._mock_programs_api(data)
data = ProgramFactory.create_batch(3)
mock_get_programs.return_value = data
program_ids = []
course_ids = []
program_uuids = []
course_run_keys = []
for program in data:
program_ids.append(program['id'])
program_uuids.append(program['uuid'])
for course_code in program['course_codes']:
for run_mode in course_code['run_modes']:
course_ids.append(run_mode['course_key'])
for course in program['courses']:
for course_run in course['course_runs']:
course_run_keys.append(course_run['key'])
# Verify that no programs are complete.
meter = utils.ProgramProgressMeter(self.user)
meter = ProgramProgressMeter(self.user)
self.assertEqual(meter.completed_programs, [])
# "Complete" all programs.
self._create_enrollments(*course_ids)
mock_get_completed_courses.return_value = [
{'course_id': course_id, 'mode': MODES.verified} for course_id in course_ids
# Complete all programs.
self._create_enrollments(*course_run_keys)
mock_completed_course_runs.return_value = [
{'course_run_id': course_run_key, 'type': MODES.verified}
for course_run_key in course_run_keys
]
# Verify that all programs are complete.
meter = utils.ProgramProgressMeter(self.user)
self.assertEqual(meter.completed_programs, program_ids)
meter = ProgramProgressMeter(self.user)
self.assertEqual(meter.completed_programs, program_uuids)
@mock.patch(UTILS_MODULE + '.certificate_api.get_certificates_for_user')
def test_completed_course_runs(self, mock_get_certificates_for_user, _mock_get_programs):
"""
Verify that the method can find course run certificates when not mocked out.
"""
def make_certificate_result(**kwargs):
"""Helper to create dummy results from the certificates API."""
result = {
'username': 'dummy-username',
'course_key': 'dummy-course',
'type': 'dummy-type',
'status': 'dummy-status',
'download_url': 'http://www.example.com/cert.pdf',
'grade': '0.98',
'created': '2015-07-31T00:00:00Z',
'modified': '2015-07-31T00:00:00Z',
}
result.update(**kwargs)
return result
mock_get_certificates_for_user.return_value = [
make_certificate_result(status='downloadable', type='verified', course_key='downloadable-course'),
make_certificate_result(status='generating', type='honor', course_key='generating-course'),
make_certificate_result(status='unknown', course_key='unknown-course'),
]
meter = ProgramProgressMeter(self.user)
self.assertEqual(
meter.completed_course_runs,
[
{'course_run_id': 'downloadable-course', 'type': 'verified'},
{'course_run_id': 'generating-course', 'type': 'honor'},
]
)
mock_get_certificates_for_user.assert_called_with(self.user.username)
@ddt.ddt
@override_settings(ECOMMERCE_PUBLIC_URL_ROOT=ECOMMERCE_URL_ROOT)
@skip_unless_lms
@mock.patch(UTILS_MODULE + '.get_run_marketing_url', mock.Mock(return_value=MARKETING_URL))
class TestProgramDataExtender(ProgramsApiConfigMixin, ModuleStoreTestCase):
class TestProgramDataExtender(ModuleStoreTestCase):
"""Tests of the program data extender utility class."""
maxDiff = None
sku = 'abc123'
......@@ -645,47 +427,47 @@ class TestProgramDataExtender(ProgramsApiConfigMixin, ModuleStoreTestCase):
self.user = UserFactory()
self.client.login(username=self.user.username, password=self.password)
ClientFactory(name=ProgramsApiConfig.OAUTH2_CLIENT_NAME, client_type=CONFIDENTIAL)
self.course = CourseFactory()
self.course = ModuleStoreCourseFactory()
self.course.start = datetime.datetime.now(utc) - datetime.timedelta(days=1)
self.course.end = datetime.datetime.now(utc) + datetime.timedelta(days=1)
self.course = self.update_course(self.course, self.user.id) # pylint: disable=no-member
self.organization = factories.Organization()
self.run_mode = factories.RunMode(course_key=unicode(self.course.id)) # pylint: disable=no-member
self.course_code = factories.CourseCode(run_modes=[self.run_mode])
self.program = factories.Program(
organizations=[self.organization],
course_codes=[self.course_code]
)
organization = OrganizationFactory()
course_run = CourseRunFactory(key=unicode(self.course.id)) # pylint: disable=no-member
course = CourseFactory(course_runs=[course_run])
program = ProgramFactory(authoring_organizations=[organization], courses=[course])
self.program = munge_catalog_program(program)
self.course_code = self.program['course_codes'][0]
self.run_mode = self.course_code['run_modes'][0]
def _assert_supplemented(self, actual, **kwargs):
"""DRY helper used to verify that program data is extended correctly."""
course_overview = CourseOverview.get_from_id(self.course.id) # pylint: disable=no-member
run_mode = dict(
factories.RunMode(
certificate_url=None,
course_image_url=course_overview.course_image_url,
course_key=unicode(self.course.id), # pylint: disable=no-member
course_url=reverse('course_root', args=[self.course.id]), # pylint: disable=no-member
end_date=self.course.end.replace(tzinfo=utc),
enrollment_open_date=strftime_localized(utils.DEFAULT_ENROLLMENT_START_DATE, 'SHORT_DATE'),
is_course_ended=self.course.end < datetime.datetime.now(utc),
is_enrolled=False,
is_enrollment_open=True,
marketing_url=MARKETING_URL,
start_date=self.course.start.replace(tzinfo=utc),
upgrade_url=None,
advertised_start=None
),
{
'certificate_url': None,
'course_image_url': course_overview.course_image_url,
'course_key': unicode(self.course.id), # pylint: disable=no-member
'course_url': reverse('course_root', args=[self.course.id]), # pylint: disable=no-member
'end_date': self.course.end.replace(tzinfo=utc),
'enrollment_open_date': strftime_localized(DEFAULT_ENROLLMENT_START_DATE, 'SHORT_DATE'),
'is_course_ended': self.course.end < datetime.datetime.now(utc),
'is_enrolled': False,
'is_enrollment_open': True,
'marketing_url': self.run_mode['marketing_url'],
'mode_slug': 'verified',
'start_date': self.course.start.replace(tzinfo=utc),
'upgrade_url': None,
'advertised_start': None,
},
**kwargs
)
course_code = factories.CourseCode(display_name=self.course_code['display_name'], run_modes=[run_mode])
expected = copy.deepcopy(self.program)
expected['course_codes'] = [course_code]
self.assertEqual(actual, expected)
self.course_code['run_modes'] = [run_mode]
self.program['course_codes'] = [self.course_code]
self.assertEqual(actual, self.program)
@ddt.data(
(False, None, False),
......@@ -711,7 +493,7 @@ class TestProgramDataExtender(ProgramsApiConfigMixin, ModuleStoreTestCase):
if is_enrolled:
CourseEnrollmentFactory(user=self.user, course_id=self.course.id, mode=enrolled_mode) # pylint: disable=no-member
data = utils.ProgramDataExtender(self.program, self.user).extend()
data = ProgramDataExtender(self.program, self.user).extend()
self._assert_supplemented(
data,
......@@ -731,7 +513,7 @@ class TestProgramDataExtender(ProgramsApiConfigMixin, ModuleStoreTestCase):
is_active=False,
)
data = utils.ProgramDataExtender(self.program, self.user).extend()
data = ProgramDataExtender(self.program, self.user).extend()
self._assert_supplemented(data)
......@@ -746,7 +528,7 @@ class TestProgramDataExtender(ProgramsApiConfigMixin, ModuleStoreTestCase):
CourseEnrollmentFactory(user=self.user, course_id=self.course.id, mode=MODES.audit) # pylint: disable=no-member
data = utils.ProgramDataExtender(self.program, self.user).extend()
data = ProgramDataExtender(self.program, self.user).extend()
self._assert_supplemented(data, is_enrolled=True, upgrade_url=None)
......@@ -764,7 +546,7 @@ class TestProgramDataExtender(ProgramsApiConfigMixin, ModuleStoreTestCase):
self.course = self.update_course(self.course, self.user.id) # pylint: disable=no-member
data = utils.ProgramDataExtender(self.program, self.user).extend()
data = ProgramDataExtender(self.program, self.user).extend()
self._assert_supplemented(
data,
......@@ -780,7 +562,7 @@ class TestProgramDataExtender(ProgramsApiConfigMixin, ModuleStoreTestCase):
self.course.enrollment_end = datetime.datetime.now(utc) - datetime.timedelta(days=1)
self.course = self.update_course(self.course, self.user.id) # pylint: disable=no-member
data = utils.ProgramDataExtender(self.program, self.user).extend()
data = ProgramDataExtender(self.program, self.user).extend()
self._assert_supplemented(
data,
......@@ -796,7 +578,7 @@ class TestProgramDataExtender(ProgramsApiConfigMixin, ModuleStoreTestCase):
mock_get_cert_data.return_value = {'uuid': test_uuid} if is_uuid_available else {}
mock_html_certs_enabled.return_value = True
data = utils.ProgramDataExtender(self.program, self.user).extend()
data = ProgramDataExtender(self.program, self.user).extend()
expected_url = reverse(
'certificates:render_cert_by_uuid',
......@@ -810,7 +592,7 @@ class TestProgramDataExtender(ProgramsApiConfigMixin, ModuleStoreTestCase):
self.course.end = datetime.datetime.now(utc) + datetime.timedelta(days=days_offset)
self.course = self.update_course(self.course, self.user.id) # pylint: disable=no-member
data = utils.ProgramDataExtender(self.program, self.user).extend()
data = ProgramDataExtender(self.program, self.user).extend()
self._assert_supplemented(data)
......@@ -824,14 +606,14 @@ class TestProgramDataExtender(ProgramsApiConfigMixin, ModuleStoreTestCase):
'logo': mock_image
}
data = utils.ProgramDataExtender(self.program, self.user).extend()
data = ProgramDataExtender(self.program, self.user).extend()
self.assertEqual(data['organizations'][0].get('img'), mock_logo_url)
@mock.patch(UTILS_MODULE + '.get_organization_by_short_name')
def test_organization_missing(self, mock_get_organization_by_short_name):
""" Verify the logo image is not set if the organizations api returns None """
mock_get_organization_by_short_name.return_value = None
data = utils.ProgramDataExtender(self.program, self.user).extend()
data = ProgramDataExtender(self.program, self.user).extend()
self.assertEqual(data['organizations'][0].get('img'), None)
@mock.patch(UTILS_MODULE + '.get_organization_by_short_name')
......@@ -841,5 +623,5 @@ class TestProgramDataExtender(ProgramsApiConfigMixin, ModuleStoreTestCase):
but the logo is not available
"""
mock_get_organization_by_short_name.return_value = {'logo': None}
data = utils.ProgramDataExtender(self.program, self.user).extend()
data = ProgramDataExtender(self.program, self.user).extend()
self.assertEqual(data['organizations'][0].get('img'), None)
openedx/core/djangoapps/programs/utils.py
# -*- coding: utf-8 -*-
"""Helper functions for working with Programs."""
from collections import defaultdict
import datetime
from urlparse import urljoin
from django.conf import settings
from django.core.urlresolvers import reverse
from django.utils.text import slugify
from django.utils.functional import cached_property
from opaque_keys.edx.keys import CourseKey
from pytz import utc
from course_modes.models import CourseMode
from lms.djangoapps.certificates import api as certificate_api
from lms.djangoapps.commerce.utils import EcommerceService
from openedx.core.djangoapps.catalog.utils import (
get_programs as get_catalog_programs,
munge_catalog_program,
get_run_marketing_url,
)
from openedx.core.djangoapps.catalog.utils import get_programs
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from openedx.core.djangoapps.programs.models import ProgramsApiConfig
from openedx.core.lib.edx_api_utils import get_edx_api_data
from student.models import CourseEnrollment
from util.date_utils import strftime_localized
......@@ -29,78 +25,6 @@ from util.organizations_helpers import get_organization_by_short_name
DEFAULT_ENROLLMENT_START_DATE = datetime.datetime(1900, 1, 1, tzinfo=utc)
def get_programs(user, program_id=None, use_catalog=False):
"""Given a user, get programs from the Programs service.
Returned value is cached depending on user permissions. Staff users making requests
against Programs will receive unpublished programs, while regular users will only receive
published programs.
Arguments:
user (User): The user to authenticate as when requesting programs.
Keyword Arguments:
program_id (int): Identifies a specific program for which to retrieve data.
Returns:
list of dict, representing programs returned by the Programs service.
dict, if a specific program is requested.
"""
if use_catalog:
programs = [munge_catalog_program(program) for program in get_catalog_programs(user)]
else:
programs_config = ProgramsApiConfig.current()
# 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
programs = get_edx_api_data(programs_config, user, 'programs', resource_id=program_id, cache_key=cache_key)
# Mix in munged MicroMasters data from the catalog.
if not program_id:
programs += [
munge_catalog_program(micromaster) for micromaster in get_catalog_programs(user, type='MicroMasters')
]
return programs
def get_programs_by_run(programs, enrollments):
"""Intersect programs and enrollments.
Builds a dictionary of program dict lists keyed by course ID. The resulting dictionary
is suitable for use in applications where programs must be filtered by the course
runs they contain (e.g., student dashboard).
Arguments:
programs (list): Containing dictionaries representing programs.
enrollments (list): Enrollments from which course IDs to key on can be extracted.
Returns:
tuple, dict of programs keyed by course ID and list of course IDs themselves
"""
programs_by_run = {}
# enrollment.course_id is really a course key (╯ಠ_ಠ）╯︵ ┻━┻
course_ids = [unicode(e.course_id) for e in enrollments]
for program in programs:
for course_code in program['course_codes']:
for run in course_code['run_modes']:
run_id = run['course_key']
if run_id in course_ids:
program_list = programs_by_run.setdefault(run_id, list())
if program not in program_list:
program_list.append(program)
# Sort programs by name for consistent presentation.
for program_list in programs_by_run.itervalues():
program_list.sort(key=lambda p: p['name'])
return programs_by_run, course_ids
def get_program_marketing_url(programs_config):
"""Build a URL to be used when linking to program details on a marketing site."""
return urljoin(settings.MKTG_URLS.get('ROOT'), programs_config.marketing_path).rstrip('/')
......@@ -118,31 +42,21 @@ def attach_program_detail_url(programs):
list, containing extended program dicts
"""
for program in programs:
base = reverse('program_details_view', kwargs={'program_id': program['id']}).rstrip('/')
slug = slugify(program['name'])
program['detail_url'] = '{base}/{slug}'.format(base=base, slug=slug)
program['detail_url'] = reverse('program_details_view', kwargs={'program_uuid': program['uuid']})
return programs
def get_completed_courses(student):
def munge_progress_map(progress_map):
"""
Determine which courses have been completed by the user.
Args:
student:
User object representing the student
Returns:
iterable of dicts with structure {'course_id': course_key, 'mode': cert_type}
Temporary utility for making progress maps look like they were built using
data from the deprecated programs service.
Clean up of this debt is tracked by ECOM-4418.
"""
all_certs = certificate_api.get_certificates_for_user(student.username)
return [
{'course_id': unicode(cert['course_key']), 'mode': cert['type']}
for cert in all_certs
if certificate_api.is_passing_status(cert['status'])
]
progress_map['id'] = progress_map.pop('uuid')
return progress_map
class ProgramProgressMeter(object):
......@@ -154,33 +68,62 @@ class ProgramProgressMeter(object):
Keyword Arguments:
enrollments (list): List of the user's enrollments.
"""
def __init__(self, user, enrollments=None, use_catalog=False):
def __init__(self, user, enrollments=None):
self.user = user
self.enrollments = enrollments
self.course_ids = None
self.course_certs = None
self.use_catalog = use_catalog
self.programs = attach_program_detail_url(get_programs(self.user, use_catalog=use_catalog))
self.enrollments = enrollments or list(CourseEnrollment.enrollments_for_user(self.user))
self.enrollments.sort(key=lambda e: e.created, reverse=True)
# enrollment.course_id is really a CourseKey (╯ಠ_ಠ）╯︵ ┻━┻
self.course_run_ids = [unicode(e.course_id) for e in self.enrollments]
def engaged_programs(self, by_run=False):
"""Derive a list of programs in which the given user is engaged.
self.programs = attach_program_detail_url(get_programs())
def invert_programs(self):
"""Intersect programs and enrollments.
Builds a dictionary of program dict lists keyed by course run ID. The
resulting dictionary is suitable in applications where programs must be
filtered by the course runs they contain (e.g., the student dashboard).
Returns:
list of program dicts, ordered by most recent enrollment,
or dict of programs, keyed by course ID.
defaultdict, programs keyed by course run ID
"""
self.enrollments = self.enrollments or list(CourseEnrollment.enrollments_for_user(self.user))
self.enrollments.sort(key=lambda e: e.created, reverse=True)
inverted_programs = defaultdict(list)
for program in self.programs:
for course in program['courses']:
for course_run in course['course_runs']:
course_run_id = course_run['key']
if course_run_id in self.course_run_ids:
program_list = inverted_programs[course_run_id]
if program not in program_list:
program_list.append(program)
programs_by_run, self.course_ids = get_programs_by_run(self.programs, self.enrollments)
# Sort programs by title for consistent presentation.
for program_list in inverted_programs.itervalues():
program_list.sort(key=lambda p: p['title'])
if by_run:
return programs_by_run
return inverted_programs
@cached_property
def engaged_programs(self):
"""Derive a list of programs in which the given user is engaged.
Returns:
list of program dicts, ordered by most recent enrollment
"""
inverted_programs = self.invert_programs()
programs = []
for course_id in self.course_ids:
for program in programs_by_run.get(course_id, []):
# Remember that these course run ids are derived from a list of
# enrollments sorted from most recent to least recent. Iterating
# over the values in inverted_programs alone won't yield a program
# ordering consistent with the user's enrollments.
for course_run_id in self.course_run_ids:
for program in inverted_programs[course_run_id]:
# Dicts aren't a hashable type, so we can't use a set. Sets also
# aren't ordered, which is important here.
if program not in programs:
programs.append(program)
......@@ -195,21 +138,23 @@ class ProgramProgressMeter(object):
towards completing a program.
"""
progress = []
for program in self.engaged_programs():
for program in self.engaged_programs:
completed, in_progress, not_started = [], [], []
for course_code in program['course_codes']:
name = course_code['display_name']
for course in program['courses']:
# TODO: What are these titles used for? If they're not used by
# the front-end, pass integer counts instead.
title = course['title']
if self._is_course_code_complete(course_code):
completed.append(name)
elif self._is_course_code_in_progress(course_code):
in_progress.append(name)
if self._is_course_complete(course):
completed.append(title)
elif self._is_course_in_progress(course):
in_progress.append(title)
else:
not_started.append(name)
not_started.append(title)
progress.append({
'id': program['id'],
'uuid': program['uuid'],
'completed': completed,
'in_progress': in_progress,
'not_started': not_started,
......@@ -222,75 +167,94 @@ class ProgramProgressMeter(object):
"""Identify programs completed by the student.
Returns:
list of int, each the ID of a completed program.
list of UUIDs, each identifying a completed program.
"""
return [program['id'] for program in self.programs if self._is_program_complete(program)]
return [program['uuid'] for program in self.programs if self._is_program_complete(program)]
def _is_program_complete(self, program):
"""Check if a user has completed a program.
A program is completed if the user has completed all nested course codes.
A program is completed if the user has completed all nested courses.
Arguments:
program (dict): Representing the program whose completion to assess.
Returns:
bool, whether the program is complete.
bool, indicating whether the program is complete.
"""
return all(self._is_course_code_complete(course_code) for course_code in program['course_codes'])
return all(self._is_course_complete(course) for course in program['courses'])
def _is_course_code_complete(self, course_code):
"""Check if a user has completed a course code.
def _is_course_complete(self, course):
"""Check if a user has completed a course.
A course code is completed if the user has earned a certificate
in the right mode for any nested run.
A course is completed if the user has earned a certificate for any of
the nested course runs.
Arguments:
course_code (dict): Containing nested run modes.
course (dict): Containing nested course runs.
Returns:
bool, whether the course code is complete.
bool, indicating whether the course is complete.
"""
self.course_certs = self.course_certs or get_completed_courses(self.user)
return any(self._parse(run_mode) in self.course_certs for run_mode in course_code['run_modes'])
def _is_course_code_in_progress(self, course_code):
"""Check if a user is in the process of completing a course code.
def reshape(course_run):
"""
Modify the structure of a course run dict to facilitate comparison
with course run certificates.
"""
return {
'course_run_id': course_run['key'],
# A course run's type is assumed to indicate which mode must be
# completed in order for the run to count towards program completion.
# This supports the same flexible program construction allowed by the
# old programs service (e.g., completion of an old honor-only run may
# count towards completion of a course in a program). This may change
# in the future to make use of the more rigid set of "applicable seat
# types" associated with each program type in the catalog.
'type': course_run['type'],
}
A user is in the process of completing a course code if they're
enrolled in the course.
return any(reshape(course_run) in self.completed_course_runs for course_run in course['course_runs'])
Arguments:
course_code (dict): Containing nested run modes.
@property
def completed_course_runs(self):
"""
Determine which course runs have been completed by the user.
Returns:
bool, whether the course code is in progress.
list of dicts, each representing a course run certificate
"""
return any(run_mode['course_key'] in self.course_ids for run_mode in course_code['run_modes'])
course_run_certificates = certificate_api.get_certificates_for_user(self.user.username)
return [
{'course_run_id': unicode(certificate['course_key']), 'type': certificate['type']}
for certificate in course_run_certificates
if certificate_api.is_passing_status(certificate['status'])
]
def _is_course_in_progress(self, course):
"""Check if a user is in the process of completing a course.
def _parse(self, run_mode):
"""Modify the structure of a run mode dict.
A user is considered to be in the process of completing a course if
they're enrolled in any of the nested course runs.
Arguments:
run_mode (dict): With `course_key` and `mode_slug` keys.
course (dict): Containing nested course runs.
Returns:
dict, with `course_id` and `mode` keys.
bool, indicating whether the course is in progress.
"""
parsed = {
'course_id': run_mode['course_key'],
'mode': run_mode['mode_slug'],
}
return parsed
return any(course_run['key'] in self.course_run_ids for course_run in course['course_runs'])
# pylint: disable=missing-docstring
class ProgramDataExtender(object):
"""Utility for extending program course codes with CourseOverview and CourseEnrollment data.
"""
Utility for extending program course codes with CourseOverview and
CourseEnrollment data.
Arguments:
program_data (dict): Representation of a program.
program_data (dict): Representation of a program. Note that this dict must
be formatted as if it was returned by the deprecated program service.
user (User): The user whose enrollments to inspect.
"""
def __init__(self, program_data, user):
......@@ -370,9 +334,6 @@ class ProgramDataExtender(object):
enrollment_end = self.course_overview.enrollment_end or datetime.datetime.max.replace(tzinfo=utc)
run_mode['is_enrollment_open'] = self.enrollment_start <= datetime.datetime.now(utc) < enrollment_end
def _attach_run_mode_marketing_url(self, run_mode):
run_mode['marketing_url'] = get_run_marketing_url(self.course_key, self.user)
def _attach_run_mode_start_date(self, run_mode):
run_mode['start_date'] = self.course_overview.start
......
openedx/core/lib/tests/test_edx_api_utils.py
"""Tests covering edX API utilities."""
# pylint: disable=missing-docstring
import json
from django.core.cache import cache
......@@ -9,9 +10,11 @@ from nose.plugins.attrib import attr
from edx_oauth2_provider.tests.factories import ClientFactory
from provider.constants import CONFIDENTIAL
from openedx.core.djangoapps.commerce.utils import ecommerce_api_client
from openedx.core.djangoapps.programs.models import ProgramsApiConfig
from openedx.core.djangoapps.programs.tests.mixins import ProgramsApiConfigMixin
from openedx.core.djangoapps.catalog.models import CatalogIntegration
from openedx.core.djangoapps.catalog.tests.mixins import CatalogIntegrationMixin
from openedx.core.djangoapps.catalog.utils import create_catalog_api_client
from openedx.core.djangoapps.credentials.models import CredentialsApiConfig
from openedx.core.djangoapps.credentials.tests.mixins import CredentialsApiConfigMixin
from openedx.core.djangolib.testing.utils import CacheIsolationTestCase, skip_unless_lms
from openedx.core.lib.edx_api_utils import get_edx_api_data
from student.tests.factories import UserFactory
......@@ -24,24 +27,21 @@ TEST_API_URL = 'http://www-internal.example.com/api'
@skip_unless_lms
@attr(shard=2)
@httpretty.activate
class TestGetEdxApiData(ProgramsApiConfigMixin, CacheIsolationTestCase):
class TestGetEdxApiData(CatalogIntegrationMixin, CredentialsApiConfigMixin, CacheIsolationTestCase):
"""Tests for edX API data retrieval utility."""
ENABLED_CACHES = ['default']
def setUp(self):
super(TestGetEdxApiData, self).setUp()
self.user = UserFactory()
ClientFactory(name=ProgramsApiConfig.OAUTH2_CLIENT_NAME, client_type=CONFIDENTIAL)
cache.clear()
def _mock_programs_api(self, responses, url=None):
"""Helper for mocking out Programs API URLs."""
self.assertTrue(httpretty.is_enabled(), msg='httpretty must be enabled to mock Programs API calls.')
def _mock_catalog_api(self, responses, url=None):
self.assertTrue(httpretty.is_enabled(), msg='httpretty must be enabled to mock Catalog API calls.')
url = url if url else ProgramsApiConfig.current().internal_api_url.strip('/') + '/programs/'
url = url if url else CatalogIntegration.current().internal_api_url.strip('/') + '/programs/'
httpretty.register_uri(httpretty.GET, url, responses=responses)
......@@ -51,7 +51,8 @@ class TestGetEdxApiData(ProgramsApiConfigMixin, CacheIsolationTestCase):
def test_get_unpaginated_data(self):
"""Verify that unpaginated data can be retrieved."""
program_config = self.create_programs_config()
catalog_integration = self.create_catalog_integration()
api = create_catalog_api_client(self.user, catalog_integration)
expected_collection = ['some', 'test', 'data']
data = {
......@@ -59,11 +60,15 @@ class TestGetEdxApiData(ProgramsApiConfigMixin, CacheIsolationTestCase):
'results': expected_collection,
}
self._mock_programs_api(
self._mock_catalog_api(
[httpretty.Response(body=json.dumps(data), content_type='application/json')]
)
actual_collection = get_edx_api_data(program_config, self.user, 'programs')
with mock.patch('openedx.core.lib.edx_api_utils.EdxRestApiClient.__init__') as mock_init:
actual_collection = get_edx_api_data(catalog_integration, self.user, 'programs', api=api)
# Verify that the helper function didn't initialize its own client.
self.assertFalse(mock_init.called)
self.assertEqual(actual_collection, expected_collection)
# Verify the API was actually hit (not the cache)
......@@ -71,10 +76,11 @@ class TestGetEdxApiData(ProgramsApiConfigMixin, CacheIsolationTestCase):
def test_get_paginated_data(self):
"""Verify that paginated data can be retrieved."""
program_config = self.create_programs_config()
catalog_integration = self.create_catalog_integration()
api = create_catalog_api_client(self.user, catalog_integration)
expected_collection = ['some', 'test', 'data']
url = ProgramsApiConfig.current().internal_api_url.strip('/') + '/programs/?page={}'
url = CatalogIntegration.current().internal_api_url.strip('/') + '/programs/?page={}'
responses = []
for page, record in enumerate(expected_collection, start=1):
......@@ -88,38 +94,40 @@ class TestGetEdxApiData(ProgramsApiConfigMixin, CacheIsolationTestCase):
httpretty.Response(body=body, content_type='application/json')
)
self._mock_programs_api(responses)
self._mock_catalog_api(responses)
actual_collection = get_edx_api_data(program_config, self.user, 'programs')
actual_collection = get_edx_api_data(catalog_integration, self.user, 'programs', api=api)
self.assertEqual(actual_collection, expected_collection)
self._assert_num_requests(len(expected_collection))
def test_get_specific_resource(self):
"""Verify that a specific resource can be retrieved."""
program_config = self.create_programs_config()
catalog_integration = self.create_catalog_integration()
api = create_catalog_api_client(self.user, catalog_integration)
resource_id = 1
url = '{api_root}/programs/{resource_id}/'.format(
api_root=ProgramsApiConfig.current().internal_api_url.strip('/'),
api_root=CatalogIntegration.current().internal_api_url.strip('/'),
resource_id=resource_id,
)
expected_resource = {'key': 'value'}
self._mock_programs_api(
self._mock_catalog_api(
[httpretty.Response(body=json.dumps(expected_resource), content_type='application/json')],
url=url
)
actual_resource = get_edx_api_data(program_config, self.user, 'programs', resource_id=resource_id)
actual_resource = get_edx_api_data(catalog_integration, self.user, 'programs', api=api, resource_id=resource_id)
self.assertEqual(actual_resource, expected_resource)
self._assert_num_requests(1)
def test_cache_utilization(self):
"""Verify that when enabled, the cache is used."""
program_config = self.create_programs_config(cache_ttl=5)
catalog_integration = self.create_catalog_integration(cache_ttl=5)
api = create_catalog_api_client(self.user, catalog_integration)
expected_collection = ['some', 'test', 'data']
data = {
......@@ -127,35 +135,37 @@ class TestGetEdxApiData(ProgramsApiConfigMixin, CacheIsolationTestCase):
'results': expected_collection,
}
self._mock_programs_api(
self._mock_catalog_api(
[httpretty.Response(body=json.dumps(data), content_type='application/json')],
)
resource_id = 1
url = '{api_root}/programs/{resource_id}/'.format(
api_root=ProgramsApiConfig.current().internal_api_url.strip('/'),
api_root=CatalogIntegration.current().internal_api_url.strip('/'),
resource_id=resource_id,
)
expected_resource = {'key': 'value'}
self._mock_programs_api(
self._mock_catalog_api(
[httpretty.Response(body=json.dumps(expected_resource), content_type='application/json')],
url=url
)
cache_key = ProgramsApiConfig.current().CACHE_KEY
cache_key = CatalogIntegration.current().CACHE_KEY
# Warm up the cache.
get_edx_api_data(program_config, self.user, 'programs', cache_key=cache_key)
get_edx_api_data(program_config, self.user, 'programs', resource_id=resource_id, cache_key=cache_key)
get_edx_api_data(catalog_integration, self.user, 'programs', api=api, cache_key=cache_key)
get_edx_api_data(
catalog_integration, self.user, 'programs', api=api, resource_id=resource_id, cache_key=cache_key
)
# Hit the cache.
actual_collection = get_edx_api_data(program_config, self.user, 'programs', cache_key=cache_key)
actual_collection = get_edx_api_data(catalog_integration, self.user, 'programs', api=api, cache_key=cache_key)
self.assertEqual(actual_collection, expected_collection)
actual_resource = get_edx_api_data(
program_config, self.user, 'programs', resource_id=resource_id, cache_key=cache_key
catalog_integration, self.user, 'programs', api=api, resource_id=resource_id, cache_key=cache_key
)
self.assertEqual(actual_resource, expected_resource)
......@@ -165,9 +175,9 @@ class TestGetEdxApiData(ProgramsApiConfigMixin, CacheIsolationTestCase):
@mock.patch(UTILITY_MODULE + '.log.warning')
def test_api_config_disabled(self, mock_warning):
"""Verify that no data is retrieved if the provided config model is disabled."""
program_config = self.create_programs_config(enabled=False)
catalog_integration = self.create_catalog_integration(enabled=False)
actual = get_edx_api_data(program_config, self.user, 'programs')
actual = get_edx_api_data(catalog_integration, self.user, 'programs')
self.assertTrue(mock_warning.called)
self.assertEqual(actual, [])
......@@ -178,9 +188,9 @@ class TestGetEdxApiData(ProgramsApiConfigMixin, CacheIsolationTestCase):
"""Verify that an exception is logged when the API client fails to initialize."""
mock_init.side_effect = Exception
program_config = self.create_programs_config()
catalog_integration = self.create_catalog_integration()
actual = get_edx_api_data(program_config, self.user, 'programs')
actual = get_edx_api_data(catalog_integration, self.user, 'programs')
self.assertTrue(mock_exception.called)
self.assertEqual(actual, [])
......@@ -188,24 +198,24 @@ class TestGetEdxApiData(ProgramsApiConfigMixin, CacheIsolationTestCase):
@mock.patch(UTILITY_MODULE + '.log.exception')
def test_data_retrieval_failure(self, mock_exception):
"""Verify that an exception is logged when data can't be retrieved."""
program_config = self.create_programs_config()
catalog_integration = self.create_catalog_integration()
api = create_catalog_api_client(self.user, catalog_integration)
self._mock_programs_api(
self._mock_catalog_api(
[httpretty.Response(body='clunk', content_type='application/json', status_code=500)]
)
actual = get_edx_api_data(program_config, self.user, 'programs')
actual = get_edx_api_data(catalog_integration, self.user, 'programs', api=api)
self.assertTrue(mock_exception.called)
self.assertEqual(actual, [])
@override_settings(ECOMMERCE_API_URL=TEST_API_URL)
def test_client_passed(self):
""" Verify that when API client is passed edx_rest_api_client is not
used.
"""
program_config = self.create_programs_config()
api = ecommerce_api_client(self.user)
def test_api_client_not_provided(self):
"""Verify that an API client doesn't need to be provided."""
ClientFactory(name=CredentialsApiConfig.OAUTH2_CLIENT_NAME, client_type=CONFIDENTIAL)
credentials_api_config = self.create_credentials_config()
with mock.patch('openedx.core.lib.edx_api_utils.EdxRestApiClient.__init__') as mock_init:
get_edx_api_data(program_config, self.user, 'orders', api=api)
self.assertFalse(mock_init.called)
get_edx_api_data(credentials_api_config, self.user, 'credentials')
self.assertTrue(mock_init.called)
pavelib/utils/envs.py
......@@ -111,11 +111,6 @@ class Env(object):
'log': BOK_CHOY_LOG_DIR / "bok_choy_ecommerce.log",
},
'programs': {
'port': 8090,
'log': BOK_CHOY_LOG_DIR / "bok_choy_programs.log",
},
'catalog': {
'port': 8091,
'log': BOK_CHOY_LOG_DIR / "bok_choy_catalog.log",
......
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