Skip to content

E
edx-platform
Commit 7a31441e by Jesse Shapiro
Options

Move to new consent API

parent b649d5a2
Showing with 551 additions and 577 deletions
cms/envs/common.py
......@@ -355,6 +355,7 @@ AUTHENTICATION_BACKENDS = (
LMS_BASE = None
LMS_ROOT_URL = "http://localhost:8000"
ENTERPRISE_API_URL = LMS_ROOT_URL + '/enterprise/api/v1/'
ENTERPRISE_CONSENT_API_URL = LMS_ROOT_URL + '/consent/api/v1/'
# These are standard regexes for pulling out info like course_ids, usage_ids, etc.
# They are used so that URLs with deprecated-format strings still work.
......@@ -1177,6 +1178,7 @@ OPTIONAL_APPS = (
# Enterprise App (http://github.com/edx/edx-enterprise)
('enterprise', None),
('consent', None),
)
......
common/djangoapps/course_modes/tests/test_views.py
......@@ -21,7 +21,6 @@ from lms.djangoapps.commerce.tests import test_utils as ecomm_test_utils
from openedx.core.djangoapps.catalog.tests.mixins import CatalogIntegrationMixin
from openedx.core.djangoapps.embargo.test_utils import restrict_course
from openedx.core.djangoapps.theming.tests.test_util import with_comprehensive_theme
from openedx.features.enterprise_support.tests.mixins.enterprise import EnterpriseServiceMockMixin
from student.models import CourseEnrollment
from student.tests.factories import CourseEnrollmentFactory, UserFactory
from util import organizations_helpers as organizations_api
......@@ -35,7 +34,7 @@ from xmodule.modulestore.tests.factories import CourseFactory
@attr(shard=3)
@ddt.ddt
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
class CourseModeViewTest(CatalogIntegrationMixin, UrlResetMixin, ModuleStoreTestCase, EnterpriseServiceMockMixin, CourseCatalogServiceMockMixin):
class CourseModeViewTest(CatalogIntegrationMixin, UrlResetMixin, ModuleStoreTestCase, CourseCatalogServiceMockMixin):
"""
Course Mode View tests
"""
......@@ -48,13 +47,6 @@ class CourseModeViewTest(CatalogIntegrationMixin, UrlResetMixin, ModuleStoreTest
self.user = UserFactory.create(username="Bob", email="bob@example.com", password="edx")
self.client.login(username=self.user.username, password="edx")
# Create a service user, because the track selection page depends on it
UserFactory.create(
username='enterprise_worker',
email="enterprise_worker@example.com",
password="edx",
)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
@httpretty.activate
@ddt.data(
......@@ -82,8 +74,6 @@ class CourseModeViewTest(CatalogIntegrationMixin, UrlResetMixin, ModuleStoreTest
user=self.user
)
self.mock_enterprise_learner_api()
# Configure whether we're upgrading or not
url = reverse('course_modes_choose', args=[unicode(self.course.id)])
response = self.client.get(url)
......@@ -133,109 +123,6 @@ class CourseModeViewTest(CatalogIntegrationMixin, UrlResetMixin, ModuleStoreTest
self.assertRedirects(response, 'http://testserver/basket/add/?sku=TEST', fetch_redirect_response=False)
ecomm_test_utils.update_commerce_config(enabled=False)
def _generate_enterprise_learner_context(self, enable_audit_enrollment=False):
"""
Internal helper to support common pieces of test case variations
"""
# Create the course modes
for mode in ('audit', 'honor', 'verified'):
CourseModeFactory.create(mode_slug=mode, course_id=self.course.id)
catalog_integration = self.create_catalog_integration()
UserFactory(username=catalog_integration.service_username)
self.mock_course_discovery_api_for_catalog_contains(
catalog_id=1, course_run_ids=[str(self.course.id)]
)
self.mock_enterprise_learner_api(enable_audit_enrollment=enable_audit_enrollment)
return reverse('course_modes_choose', args=[unicode(self.course.id)])
@httpretty.activate
def test_no_enrollment(self):
url = self._generate_enterprise_learner_context()
response = self.client.get(url)
self.assertEquals(response.status_code, 200)
@httpretty.activate
@waffle.testutils.override_switch("populate-multitenant-programs", True)
def test_enterprise_learner_context(self):
"""
Test: Track selection page should show the enterprise context message if user belongs to the Enterprise.
"""
url = self._generate_enterprise_learner_context()
# User visits the track selection page directly without ever enrolling
response = self.client.get(url)
self.assertEquals(response.status_code, 200)
self.assertContains(
response,
'Welcome, {username}! You are about to enroll in {course_name}, from {partner_names}, '
'sponsored by TestShib. Please select your enrollment information below.'.format(
username=self.user.username,
course_name=self.course.display_name_with_default_escaped,
partner_names=self.course.org
)
)
@httpretty.activate
@waffle.testutils.override_switch("populate-multitenant-programs", True)
def test_enterprise_learner_context_with_multiple_organizations(self):
"""
Test: Track selection page should show the enterprise context message with multiple organization names
if user belongs to the Enterprise.
"""
url = self._generate_enterprise_learner_context()
# Creating organization
for i in xrange(2):
test_organization_data = {
'name': 'test organization ' + str(i),
'short_name': 'test_organization_' + str(i),
'description': 'Test Organization Description',
'active': True,
'logo': '/logo_test1.png/'
}
test_org = organizations_api.add_organization(organization_data=test_organization_data)
organizations_api.add_organization_course(organization_data=test_org, course_id=unicode(self.course.id))
# User visits the track selection page directly without ever enrolling
response = self.client.get(url)
self.assertEquals(response.status_code, 200)
self.assertContains(
response,
'Welcome, {username}! You are about to enroll in {course_name}, from test organization 0 and '
'test organization 1, sponsored by TestShib. Please select your enrollment information below.'.format(
username=self.user.username,
course_name=self.course.display_name_with_default_escaped
)
)
@httpretty.activate
@waffle.testutils.override_switch("populate-multitenant-programs", True)
def test_enterprise_learner_context_audit_disabled(self):
"""
Track selection page should hide the audit choice by default in an Enterprise Customer/Learner context
"""
# User visits the track selection page directly without ever enrolling, sees only Verified track choice
url = self._generate_enterprise_learner_context()
response = self.client.get(url)
self.assertContains(response, 'Pursue a Verified Certificate')
self.assertNotContains(response, 'Audit This Course')
@httpretty.activate
def test_enterprise_learner_context_audit_enabled(self):
"""
Track selection page should display Audit choice when specified for an Enterprise Customer
"""
# User visits the track selection page directly without ever enrolling, sees both Verified and Audit choices
url = self._generate_enterprise_learner_context(enable_audit_enrollment=True)
response = self.client.get(url)
self.assertContains(response, 'Pursue a Verified Certificate')
self.assertContains(response, 'Audit This Course')
@httpretty.activate
@ddt.data(
'',
......@@ -263,8 +150,6 @@ class CourseModeViewTest(CatalogIntegrationMixin, UrlResetMixin, ModuleStoreTest
user=self.user
)
self.mock_enterprise_learner_api()
# Verify that the prices render correctly
response = self.client.get(
reverse('course_modes_choose', args=[unicode(self.course.id)]),
......@@ -286,8 +171,6 @@ class CourseModeViewTest(CatalogIntegrationMixin, UrlResetMixin, ModuleStoreTest
for mode in available_modes:
CourseModeFactory.create(mode_slug=mode, course_id=self.course.id)
self.mock_enterprise_learner_api()
# Check whether credit upsell is shown on the page
# This should *only* be shown when a credit mode is available
url = reverse('course_modes_choose', args=[unicode(self.course.id)])
......@@ -530,8 +413,6 @@ class CourseModeViewTest(CatalogIntegrationMixin, UrlResetMixin, ModuleStoreTest
for mode in ["honor", "verified"]:
CourseModeFactory.create(mode_slug=mode, course_id=self.course.id)
self.mock_enterprise_learner_api()
# Load the track selection page
url = reverse('course_modes_choose', args=[unicode(self.course.id)])
response = self.client.get(url)
......@@ -558,7 +439,7 @@ class CourseModeViewTest(CatalogIntegrationMixin, UrlResetMixin, ModuleStoreTest
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
class TrackSelectionEmbargoTest(UrlResetMixin, ModuleStoreTestCase, EnterpriseServiceMockMixin):
class TrackSelectionEmbargoTest(UrlResetMixin, ModuleStoreTestCase):
"""Test embargo restrictions on the track selection page. """
URLCONF_MODULES = ['openedx.core.djangoapps.embargo']
......@@ -576,13 +457,6 @@ class TrackSelectionEmbargoTest(UrlResetMixin, ModuleStoreTestCase, EnterpriseSe
self.user = UserFactory.create(username="Bob", email="bob@example.com", password="edx")
self.client.login(username=self.user.username, password="edx")
# Create a service user
UserFactory.create(
username='enterprise_worker',
email="enterprise_worker@example.com",
password="edx",
)
# Construct the URL for the track selection page
self.url = reverse('course_modes_choose', args=[unicode(self.course.id)])
......@@ -595,6 +469,5 @@ class TrackSelectionEmbargoTest(UrlResetMixin, ModuleStoreTestCase, EnterpriseSe
@httpretty.activate
def test_embargo_allow(self):
self.mock_enterprise_learner_api()
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
common/djangoapps/course_modes/views.py
......@@ -24,7 +24,6 @@ from edxmako.shortcuts import render_to_response
from lms.djangoapps.commerce.utils import EcommerceService
from lms.djangoapps.experiments.utils import get_experiment_user_metadata_context
from openedx.core.djangoapps.embargo import api as embargo_api
from openedx.features.enterprise_support import api as enterprise_api
from student.models import CourseEnrollment
from third_party_auth.decorators import tpa_hint_ends_existing_session
from util import organizations_helpers as organization_api
......@@ -162,36 +161,6 @@ class ChooseModeView(View):
title_content = _("Congratulations! You are now enrolled in {course_name}").format(
course_name=course.display_name_with_default_escaped
)
enterprise_learner_data = enterprise_api.get_enterprise_learner_data(site=request.site, user=request.user)
if enterprise_learner_data:
enterprise_learner = enterprise_learner_data[0]
is_course_in_enterprise_catalog = enterprise_api.is_course_in_enterprise_catalog(
site=request.site,
course_id=course_id,
enterprise_catalog_id=enterprise_learner['enterprise_customer']['catalog']
)
if is_course_in_enterprise_catalog:
partner_names = partner_name = course.display_organization \
if course.display_organization else course.org
enterprise_name = enterprise_learner['enterprise_customer']['name']
organizations = organization_api.get_course_organizations(course_id=course.id)
if organizations:
partner_names = ' and '.join([org.get('name', partner_name) for org in organizations])
title_content = _("Welcome, {username}! You are about to enroll in {course_name},"
" from {partner_names}, sponsored by {enterprise_name}. Please select your enrollment"
" information below.").format(
username=request.user.username,
course_name=course.display_name_with_default_escaped,
partner_names=partner_names,
enterprise_name=enterprise_name
)
# Hide the audit modes for this enterprise customer, if necessary
if not enterprise_learner['enterprise_customer'].get('enable_audit_enrollment'):
for audit_mode in CourseMode.AUDIT_MODES:
modes.pop(audit_mode, None)
context["title_content"] = title_content
......
common/djangoapps/enrollment/tests/test_views.py
......@@ -56,6 +56,7 @@ class EnrollmentTestMixin(object):
min_mongo_calls=0,
max_mongo_calls=0,
enterprise_course_consent=None,
linked_enterprise_customer=None,
):
"""
Enroll in the course and verify the response's status code. If the expected status is 200, also validates
......@@ -85,6 +86,9 @@ class EnrollmentTestMixin(object):
if enterprise_course_consent is not None:
data['enterprise_course_consent'] = enterprise_course_consent
if linked_enterprise_customer is not None:
data['linked_enterprise_customer'] = linked_enterprise_customer
extra = {}
if as_server:
extra['HTTP_X_EDX_API_KEY'] = self.API_KEY
......@@ -961,6 +965,7 @@ class EnrollmentTest(EnrollmentTestMixin, ModuleStoreTestCase, APITestCase, Ente
self.assertTrue(is_active)
self.assertEqual(course_mode, updated_mode)
@override_settings(ENABLE_ENTERPRISE_INTEGRATION=True)
def test_enterprise_course_enrollment_invalid_consent(self):
"""Verify that the enterprise_course_consent must be a boolean. """
CourseModeFactory.create(
......@@ -976,6 +981,7 @@ class EnrollmentTest(EnrollmentTestMixin, ModuleStoreTestCase, APITestCase, Ente
@httpretty.activate
@override_settings(ENTERPRISE_SERVICE_WORKER_USERNAME='enterprise_worker')
@override_settings(ENABLE_ENTERPRISE_INTEGRATION=True)
def test_enterprise_course_enrollment_api_error(self):
"""Verify that enterprise service errors are handled properly. """
UserFactory.create(
......@@ -1003,6 +1009,7 @@ class EnrollmentTest(EnrollmentTestMixin, ModuleStoreTestCase, APITestCase, Ente
@httpretty.activate
@override_settings(ENTERPRISE_SERVICE_WORKER_USERNAME='enterprise_worker')
@override_settings(ENABLE_ENTERPRISE_INTEGRATION=True)
def test_enterprise_course_enrollment_successful(self):
"""Verify that the enrollment completes when the EnterpriseCourseEnrollment creation succeeds. """
UserFactory.create(
......@@ -1028,6 +1035,43 @@ class EnrollmentTest(EnrollmentTestMixin, ModuleStoreTestCase, APITestCase, Ente
'No request was made to the mocked enterprise-course-enrollment API'
)
@httpretty.activate
@override_settings(ENTERPRISE_SERVICE_WORKER_USERNAME='enterprise_worker')
@override_settings(ENABLE_ENTERPRISE_INTEGRATION=True)
def test_enterprise_course_enrollment_with_ec_uuid(self):
"""Verify that the enrollment completes when the EnterpriseCourseEnrollment creation succeeds. """
UserFactory.create(
username='enterprise_worker',
email=self.EMAIL,
password=self.PASSWORD,
)
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=CourseMode.DEFAULT_MODE_SLUG,
mode_display_name=CourseMode.DEFAULT_MODE_SLUG,
)
consent_kwargs = {
'username': self.user.username,
'course_id': unicode(self.course.id),
'ec_uuid': 'this-is-a-real-uuid'
}
self.mock_consent_missing(**consent_kwargs)
self.mock_consent_post(**consent_kwargs)
self.assert_enrollment_status(
expected_status=status.HTTP_200_OK,
as_server=True,
username='enterprise_worker',
linked_enterprise_customer='this-is-a-real-uuid',
)
self.assertEqual(
httpretty.last_request().path,
'/consent/api/v1/data_sharing_consent',
)
self.assertEqual(
httpretty.last_request().method,
httpretty.POST
)
def test_enrollment_attributes_always_written(self):
""" Enrollment attributes should always be written, regardless of whether
the enrollment is being created or updated.
......
common/djangoapps/enrollment/views.py
......@@ -29,7 +29,12 @@ from openedx.core.lib.api.authentication import (
from openedx.core.lib.api.permissions import ApiKeyHeaderPermission, ApiKeyHeaderPermissionIsAuthenticated
from openedx.core.lib.exceptions import CourseNotFoundError
from openedx.core.lib.log_utils import audit_log
from openedx.features.enterprise_support.api import EnterpriseApiClient, EnterpriseApiException, enterprise_enabled
from openedx.features.enterprise_support.api import (
ConsentApiClient,
EnterpriseApiClient,
EnterpriseApiException,
enterprise_enabled
)
from student.auth import user_has_role
from student.models import User
from student.roles import CourseStaffRole, GlobalStaff
......@@ -591,27 +596,42 @@ class EnrollmentListView(APIView, ApiKeyPermissionMixIn):
)
enterprise_course_consent = request.data.get('enterprise_course_consent')
# Check if the enterprise_course_enrollment is a boolean
if has_api_key_permissions and enterprise_enabled() and enterprise_course_consent is not None:
if not isinstance(enterprise_course_consent, bool):
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={
'message': (u"'{value}' is an invalid enterprise course consent value.").format(
value=enterprise_course_consent
)
}
)
try:
EnterpriseApiClient().post_enterprise_course_enrollment(
username,
unicode(course_id),
enterprise_course_consent
)
except EnterpriseApiException as error:
log.exception("An unexpected error occurred while creating the new EnterpriseCourseEnrollment "
"for user [%s] in course run [%s]", username, course_id)
raise CourseEnrollmentError(error.message)
explicit_linked_enterprise = request.data.get('linked_enterprise_customer')
if has_api_key_permissions and enterprise_enabled():
# We received an explicitly-linked EnterpriseCustomer for the enrollment
if explicit_linked_enterprise is not None:
kwargs = {
'username': username,
'course_id': unicode(course_id),
'enterprise_customer_uuid': explicit_linked_enterprise,
}
consent_client = ConsentApiClient()
consent_client.provide_consent(**kwargs)
# We received an implicit "consent granted" parameter from ecommerce
# TODO: Once ecommerce has been deployed with explicit enterprise support, remove this
# entire chunk of logic, related tests, and any supporting methods no longer required.
elif enterprise_course_consent is not None:
# Check if the enterprise_course_enrollment is a boolean
if not isinstance(enterprise_course_consent, bool):
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={
'message': (u"'{value}' is an invalid enterprise course consent value.").format(
value=enterprise_course_consent
)
}
)
try:
EnterpriseApiClient().post_enterprise_course_enrollment(
username,
unicode(course_id),
enterprise_course_consent
)
except EnterpriseApiException as error:
log.exception("An unexpected error occurred while creating the new EnterpriseCourseEnrollment "
"for user [%s] in course run [%s]", username, course_id)
raise CourseEnrollmentError(error.message)
enrollment_attributes = request.data.get('enrollment_attributes')
enrollment = api.get_enrollment(username, unicode(course_id))
......
lms/djangoapps/courseware/tests/test_course_info.py
......@@ -12,6 +12,7 @@ from lms.djangoapps.ccx.tests.factories import CcxFactory
from nose.plugins.attrib import attr
from openedx.core.djangoapps.self_paced.models import SelfPacedConfiguration
from openedx.core.djangoapps.waffle_utils.testutils import WAFFLE_TABLES
from openedx.features.enterprise_support.tests.mixins.enterprise import EnterpriseTestConsentRequired
from pyquery import PyQuery as pq
from student.models import CourseEnrollment
from student.tests.factories import AdminFactory
......@@ -32,7 +33,7 @@ QUERY_COUNT_TABLE_BLACKLIST = WAFFLE_TABLES
@attr(shard=1)
class CourseInfoTestCase(LoginEnrollmentTestCase, SharedModuleStoreTestCase):
class CourseInfoTestCase(EnterpriseTestConsentRequired, LoginEnrollmentTestCase, SharedModuleStoreTestCase):
"""
Tests for the Course Info page
"""
......@@ -61,8 +62,7 @@ class CourseInfoTestCase(LoginEnrollmentTestCase, SharedModuleStoreTestCase):
self.assertNotIn("You are not currently enrolled in this course", resp.content)
# TODO: LEARNER-611: If this is only tested under Course Info, does this need to move?
@mock.patch('openedx.features.enterprise_support.api.get_enterprise_consent_url')
def test_redirection_missing_enterprise_consent(self, mock_get_url):
def test_redirection_missing_enterprise_consent(self):
"""
Verify that users viewing the course info who are enrolled, but have not provided
data sharing consent, are first redirected to a consent page, and then, once they've
......@@ -70,19 +70,10 @@ class CourseInfoTestCase(LoginEnrollmentTestCase, SharedModuleStoreTestCase):
"""
self.setup_user()
self.enroll(self.course)
mock_get_url.return_value = reverse('dashboard')
url = reverse('info', args=[self.course.id.to_deprecated_string()])
response = self.client.get(url)
url = reverse('info', args=[self.course.id.to_deprecated_string()])
self.assertRedirects(
response,
reverse('dashboard')
)
mock_get_url.assert_called_once()
mock_get_url.return_value = None
response = self.client.get(url)
self.assertNotIn("You are not currently enrolled in this course", response.content)
self.verify_consent_required(self.client, url)
def test_anonymous_user(self):
url = reverse('info', args=[self.course.id.to_deprecated_string()])
......
lms/djangoapps/courseware/tests/test_view_authentication.py
......@@ -15,6 +15,7 @@ from courseware.tests.factories import (
StaffFactory
)
from courseware.tests.helpers import CourseAccessTestMixin, LoginEnrollmentTestCase
from openedx.features.enterprise_support.tests.mixins.enterprise import EnterpriseTestConsentRequired
from student.tests.factories import CourseEnrollmentFactory, UserFactory
from xmodule.modulestore.django import modulestore
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
......@@ -22,7 +23,7 @@ from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
@attr(shard=1)
class TestViewAuth(ModuleStoreTestCase, LoginEnrollmentTestCase):
class TestViewAuth(EnterpriseTestConsentRequired, ModuleStoreTestCase, LoginEnrollmentTestCase):
"""
Check that view authentication works properly.
"""
......@@ -201,28 +202,18 @@ class TestViewAuth(ModuleStoreTestCase, LoginEnrollmentTestCase):
)
)
@patch('openedx.features.enterprise_support.api.get_enterprise_consent_url')
def test_redirection_missing_enterprise_consent(self, mock_get_url):
def test_redirection_missing_enterprise_consent(self):
"""
Verify that enrolled students are redirected to the Enterprise consent
URL if a linked Enterprise Customer requires data sharing consent
and it has not yet been provided.
"""
mock_get_url.return_value = reverse('dashboard')
self.login(self.enrolled_user)
url = reverse(
'courseware',
kwargs={'course_id': self.course.id.to_deprecated_string()}
)
response = self.client.get(url)
self.assertRedirects(
response,
reverse('dashboard')
)
mock_get_url.assert_called_once()
mock_get_url.return_value = None
response = self.client.get(url)
self.assertNotIn("You are not currently enrolled in this course", response.content)
self.verify_consent_required(self.client, url, status_code=302)
def test_instructor_page_access_nonstaff(self):
"""
......
lms/djangoapps/courseware/tests/test_views.py
......@@ -212,8 +212,8 @@ class IndexQueryTestCase(ModuleStoreTestCase):
NUM_PROBLEMS = 20
@ddt.data(
(ModuleStoreEnum.Type.mongo, 10, 146),
(ModuleStoreEnum.Type.split, 4, 146),
(ModuleStoreEnum.Type.mongo, 10, 145),
(ModuleStoreEnum.Type.split, 4, 145),
)
@ddt.unpack
def test_index_query_counts(self, store_type, expected_mongo_query_count, expected_mysql_query_count):
......
lms/djangoapps/student_account/test/test_views.py
......@@ -472,34 +472,24 @@ class StudentAccountLoginAndRegistrationTest(ThirdPartyAuthTestMixin, UrlResetMi
@mock.patch('student_account.views.enterprise_customer_for_request')
@ddt.data(
('signin_user', False, None, None, None),
('register_user', False, None, None, None),
('signin_user', True, 'Fake EC', 'http://logo.com/logo.jpg', u'{enterprise_name} - {platform_name}'),
('register_user', True, 'Fake EC', 'http://logo.com/logo.jpg', u'{enterprise_name} - {platform_name}'),
('signin_user', True, 'Fake EC', None, u'{enterprise_name} - {platform_name}'),
('register_user', True, 'Fake EC', None, u'{enterprise_name} - {platform_name}'),
('signin_user', True, 'Fake EC', 'http://logo.com/logo.jpg', None),
('register_user', True, 'Fake EC', 'http://logo.com/logo.jpg', None),
('signin_user', True, 'Fake EC', None, None),
('register_user', True, 'Fake EC', None, None),
('signin_user', False, None, None),
('register_user', False, None, None),
('signin_user', True, 'Fake EC', 'http://logo.com/logo.jpg'),
('register_user', True, 'Fake EC', 'http://logo.com/logo.jpg'),
('signin_user', True, 'Fake EC', None),
('register_user', True, 'Fake EC', None),
)
@ddt.unpack
def test_enterprise_register(self, url_name, ec_present, ec_name, logo_url, welcome_message, mock_get_ec):
def test_enterprise_register(self, url_name, ec_present, ec_name, logo_url, mock_get_ec):
"""
Verify that when an EnterpriseCustomer is received on the login and register views,
the appropriate sidebar is rendered.
"""
if ec_present:
mock_ec = mock_get_ec.return_value
mock_ec.name = ec_name
if logo_url:
mock_ec.branding_configuration.logo.url = logo_url
else:
mock_ec.branding_configuration.logo = None
if welcome_message:
mock_ec.branding_configuration.welcome_message = welcome_message
else:
del mock_ec.branding_configuration.welcome_message
mock_get_ec.return_value = {
'name': ec_name,
'branding_configuration': {'logo': logo_url}
}
else:
mock_get_ec.return_value = None
......@@ -511,8 +501,7 @@ class StudentAccountLoginAndRegistrationTest(ThirdPartyAuthTestMixin, UrlResetMi
self.assertNotContains(response, text=enterprise_sidebar_div_id)
else:
self.assertContains(response, text=enterprise_sidebar_div_id)
if not welcome_message:
welcome_message = settings.ENTERPRISE_SPECIFIC_BRANDED_WELCOME_TEMPLATE
welcome_message = settings.ENTERPRISE_SPECIFIC_BRANDED_WELCOME_TEMPLATE
expected_message = welcome_message.format(
start_bold=u'<b>',
end_bold=u'</b>',
......
lms/djangoapps/student_account/views.py
......@@ -263,23 +263,17 @@ def enterprise_sidebar_context(request):
platform_name = configuration_helpers.get_value('PLATFORM_NAME', settings.PLATFORM_NAME)
if enterprise_customer.branding_configuration.logo:
enterprise_logo_url = enterprise_customer.branding_configuration.logo.url
else:
enterprise_logo_url = ''
logo_url = enterprise_customer.get('branding_configuration', {}).get('logo', '')
if getattr(enterprise_customer.branding_configuration, 'welcome_message', None):
branded_welcome_template = enterprise_customer.branding_configuration.welcome_message
else:
branded_welcome_template = configuration_helpers.get_value(
'ENTERPRISE_SPECIFIC_BRANDED_WELCOME_TEMPLATE',
settings.ENTERPRISE_SPECIFIC_BRANDED_WELCOME_TEMPLATE
)
branded_welcome_template = configuration_helpers.get_value(
'ENTERPRISE_SPECIFIC_BRANDED_WELCOME_TEMPLATE',
settings.ENTERPRISE_SPECIFIC_BRANDED_WELCOME_TEMPLATE
)
branded_welcome_string = branded_welcome_template.format(
start_bold=u'<b>',
end_bold=u'</b>',
enterprise_name=enterprise_customer.name,
enterprise_name=enterprise_customer['name'],
platform_name=platform_name
)
......@@ -290,8 +284,8 @@ def enterprise_sidebar_context(request):
platform_welcome_string = platform_welcome_template.format(platform_name=platform_name)
context = {
'enterprise_name': enterprise_customer.name,
'enterprise_logo_url': enterprise_logo_url,
'enterprise_name': enterprise_customer['name'],
'enterprise_logo_url': logo_url,
'enterprise_branded_welcome_string': branded_welcome_string,
'platform_welcome_string': platform_welcome_string,
}
......
lms/envs/aws.py
......@@ -969,6 +969,11 @@ if LMS_ROOT_URL is not None:
DEFAULT_ENTERPRISE_API_URL = LMS_ROOT_URL + '/enterprise/api/v1/'
ENTERPRISE_API_URL = ENV_TOKENS.get('ENTERPRISE_API_URL', DEFAULT_ENTERPRISE_API_URL)
DEFAULT_ENTERPRISE_CONSENT_API_URL = None
if LMS_ROOT_URL is not None:
DEFAULT_ENTERPRISE_CONSENT_API_URL = LMS_ROOT_URL + '/consent/api/v1/'
ENTERPRISE_CONSENT_API_URL = ENV_TOKENS.get('ENTERPRISE_CONSENT_API_URL', DEFAULT_ENTERPRISE_CONSENT_API_URL)
ENTERPRISE_SERVICE_WORKER_USERNAME = ENV_TOKENS.get(
'ENTERPRISE_SERVICE_WORKER_USERNAME',
ENTERPRISE_SERVICE_WORKER_USERNAME
......
lms/envs/common.py
......@@ -3216,6 +3216,7 @@ ENTERPRISE_COURSE_ENROLLMENT_AUDIT_MODES = ['audit', 'honor']
# and are overridden by the configuration parameter accessors defined in aws.py
ENTERPRISE_API_URL = LMS_ROOT_URL + '/enterprise/api/v1/'
ENTERPRISE_CONSENT_API_URL = LMS_ROOT_URL + '/consent/api/v1/'
ENTERPRISE_SERVICE_WORKER_USERNAME = 'enterprise_worker'
ENTERPRISE_API_CACHE_TIMEOUT = 3600 # Value is in seconds
ENTERPRISE_CUSTOMER_LOGO_IMAGE_SIZE = 512 # Enterprise logo image size limit in KB's
......
lms/envs/test.py
......@@ -605,7 +605,9 @@ COMPREHENSIVE_THEME_LOCALE_PATHS = [REPO_ROOT / "themes/conf/locale", ]
LMS_ROOT_URL = "http://localhost:8000"
ENABLE_ENTERPRISE_INTEGRATION = False
ECOMMERCE_API_URL = 'https://ecommerce.example.com/api/v2/'
ENTERPRISE_API_URL = 'http://enterprise.example.com/enterprise/api/v1/'
ENTERPRISE_CONSENT_API_URL = 'http://enterprise.example.com/consent/api/v1/'
ACTIVATION_EMAIL_FROM_ADDRESS = 'test_activate@edx.org'
openedx/features/course_experience/tests/views/test_course_home.py
......@@ -160,7 +160,7 @@ class TestCourseHomePage(CourseHomePageTestCase):
course_home_url(self.course)
# Fetch the view and verify the query counts
with self.assertNumQueries(42, table_blacklist=QUERY_COUNT_TABLE_BLACKLIST):
with self.assertNumQueries(41, table_blacklist=QUERY_COUNT_TABLE_BLACKLIST):
with check_mongo_calls(4):
url = course_home_url(self.course)
self.client.get(url)
......
openedx/features/course_experience/tests/views/test_course_updates.py
......@@ -127,7 +127,7 @@ class TestCourseUpdatesPage(SharedModuleStoreTestCase):
course_updates_url(self.course)
# Fetch the view and verify that the query counts haven't changed
with self.assertNumQueries(33, table_blacklist=QUERY_COUNT_TABLE_BLACKLIST):
with self.assertNumQueries(32, table_blacklist=QUERY_COUNT_TABLE_BLACKLIST):
with check_mongo_calls(4):
url = course_updates_url(self.course)
self.client.get(url)
openedx/features/enterprise_support/api.py
......@@ -16,7 +16,7 @@ from django.utils.http import urlencode
from django.utils.translation import ugettext as _
from edx_rest_api_client.client import EdxRestApiClient
from requests.exceptions import ConnectionError, Timeout
from slumber.exceptions import HttpClientError, HttpServerError, SlumberBaseException
from slumber.exceptions import HttpClientError, HttpNotFoundError, HttpServerError, SlumberBaseException
from openedx.core.djangoapps.catalog.models import CatalogIntegration
from openedx.core.djangoapps.catalog.utils import create_catalog_api_client
......@@ -26,9 +26,7 @@ from third_party_auth.pipeline import get as get_partial_pipeline
from third_party_auth.provider import Registry
try:
from enterprise import utils as enterprise_utils
from enterprise.models import EnterpriseCourseEnrollment, EnterpriseCustomer
from enterprise.utils import consent_necessary_for_course
except ImportError:
pass
......@@ -43,6 +41,62 @@ class EnterpriseApiException(Exception):
pass
class ConsentApiClient(object):
"""
Class for producing an Enterprise Consent service API client
"""
def __init__(self):
"""
Initialize a consent service API client, authenticated using the Enterprise worker username.
"""
self.user = User.objects.get(username=settings.ENTERPRISE_SERVICE_WORKER_USERNAME)
jwt = JwtBuilder(self.user).build_token([])
url = configuration_helpers.get_value('ENTERPRISE_CONSENT_API_URL', settings.ENTERPRISE_CONSENT_API_URL)
self.client = EdxRestApiClient(
url,
jwt=jwt,
append_slash=False,
)
self.consent_endpoint = self.client.data_sharing_consent
def revoke_consent(self, **kwargs):
"""
Revoke consent from any existing records that have it at the given scope.
This endpoint takes any given kwargs, which are understood as filtering the
conceptual scope of the consent involved in the request.
"""
return self.consent_endpoint.delete(**kwargs)
def provide_consent(self, **kwargs):
"""
Provide consent at the given scope.
This endpoint takes any given kwargs, which are understood as filtering the
conceptual scope of the consent involved in the request.
"""
return self.consent_endpoint.post(kwargs)
def consent_required(self, enrollment_exists=False, **kwargs):
"""
Determine if consent is required at the given scope.
This endpoint takes any given kwargs, which are understood as filtering the
conceptual scope of the consent involved in the request.
"""
# Call the endpoint with the given kwargs, and check the value that it provides.
response = self.consent_endpoint.get(**kwargs)
# No Enterprise record exists, but we're already enrolled in a course. So, go ahead and proceed.
if enrollment_exists and not response.get('exists', False):
return False
# In all other cases, just trust the Consent API.
return response['consent_required']
class EnterpriseApiClient(object):
"""
Class for producing an Enterprise service API client.
......@@ -59,6 +113,10 @@ class EnterpriseApiClient(object):
jwt=jwt
)
def get_enterprise_customer(self, uuid):
endpoint = getattr(self.client, 'enterprise-customer')
return endpoint(uuid).get()
def post_enterprise_course_enrollment(self, username, course_id, consent_granted):
"""
Create an EnterpriseCourseEnrollment by using the corresponding serializer (for validation).
......@@ -166,25 +224,16 @@ class EnterpriseApiClient(object):
api_resource_name = 'enterprise-learner'
cache_key = get_cache_key(
site_domain=site.domain,
resource=api_resource_name,
username=user.username
)
response = cache.get(cache_key)
if not response:
try:
endpoint = getattr(self.client, api_resource_name)
querystring = {'username': user.username}
response = endpoint().get(**querystring)
cache.set(cache_key, response, settings.ENTERPRISE_API_CACHE_TIMEOUT)
except (HttpClientError, HttpServerError):
message = ("An error occurred while getting EnterpriseLearner data for user {username}".format(
username=user.username
))
LOGGER.exception(message)
return None
try:
endpoint = getattr(self.client, api_resource_name)
querystring = {'username': user.username}
response = endpoint().get(**querystring)
except (HttpClientError, HttpServerError):
message = ("An error occurred while getting EnterpriseLearner data for user {username}".format(
username=user.username
))
LOGGER.exception(message)
return None
return response
......@@ -210,7 +259,7 @@ def data_sharing_consent_required(view_func):
Otherwise, just call the wrapped view function.
"""
# Redirect to the consent URL, if consent is required.
consent_url = get_enterprise_consent_url(request, course_id)
consent_url = get_enterprise_consent_url(request, course_id, enrollment_exists=True)
if consent_url:
real_user = getattr(request.user, 'real_user', request.user)
LOGGER.warning(
......@@ -233,52 +282,98 @@ def enterprise_enabled():
return 'enterprise' in settings.INSTALLED_APPS and getattr(settings, 'ENABLE_ENTERPRISE_INTEGRATION', True)
def enterprise_customer_for_request(request, tpa_hint=None):
def enterprise_customer_for_request(request):
"""
Check all the context clues of the request to determine if
the request being made is tied to a particular EnterpriseCustomer.
"""
if not enterprise_enabled():
return None
ec = None
sso_provider_id = request.GET.get('tpa_hint')
running_pipeline = get_partial_pipeline(request)
if running_pipeline:
# Determine if the user is in the middle of a third-party auth pipeline,
# and set the tpa_hint parameter to match if so.
tpa_hint = Registry.get_from_pipeline(running_pipeline).provider_id
# and set the sso_provider_id parameter to match if so.
sso_provider_id = Registry.get_from_pipeline(running_pipeline).provider_id
if tpa_hint:
if sso_provider_id:
# If we have a third-party auth provider, get the linked enterprise customer.
try:
ec = EnterpriseCustomer.objects.get(enterprise_customer_identity_provider__provider_id=tpa_hint)
# FIXME: Implement an Enterprise API endpoint where we can get the EC
# directly via the linked SSO provider
# Check if there's an Enterprise Customer such that the linked SSO provider
# has an ID equal to the ID we got from the running pipeline or from the
# request tpa_hint URL parameter.
ec_uuid = EnterpriseCustomer.objects.get(
enterprise_customer_identity_provider__provider_id=sso_provider_id
).uuid
except EnterpriseCustomer.DoesNotExist:
pass
ec_uuid = request.GET.get('enterprise_customer') or request.COOKIES.get(settings.ENTERPRISE_CUSTOMER_COOKIE_NAME)
# If we haven't obtained an EnterpriseCustomer through the other methods, check the
# session cookies and URL parameters for an explicitly-passed EnterpriseCustomer.
if not ec and ec_uuid:
# If there is not an EnterpriseCustomer linked to this SSO provider, set
# the UUID variable to be null.
ec_uuid = None
else:
# Check if we got an Enterprise UUID passed directly as either a query
# parameter, or as a value in the Enterprise cookie.
ec_uuid = request.GET.get('enterprise_customer') or request.COOKIES.get(settings.ENTERPRISE_CUSTOMER_COOKIE_NAME)
if not ec_uuid and request.user.is_authenticated():
# If there's no way to get an Enterprise UUID for the request, check to see
# if there's already an Enterprise attached to the requesting user on the backend.
learner_data = get_enterprise_learner_data(request.site, request.user)
if learner_data:
ec_uuid = learner_data[0]['enterprise_customer']['uuid']
if ec_uuid:
# If we were able to obtain an EnterpriseCustomer UUID, go ahead
# and use it to attempt to retrieve EnterpriseCustomer details
# from the EnterpriseCustomer API.
try:
ec = EnterpriseCustomer.objects.get(uuid=ec_uuid)
except (EnterpriseCustomer.DoesNotExist, ValueError):
ec = EnterpriseApiClient().get_enterprise_customer(ec_uuid)
except HttpNotFoundError:
ec = None
return ec
def consent_needed_for_course(user, course_id):
def consent_needed_for_course(request, user, course_id, enrollment_exists=False):
"""
Wrap the enterprise app check to determine if the user needs to grant
data sharing permissions before accessing a course.
"""
if not enterprise_enabled():
return False
return consent_necessary_for_course(user, course_id)
consent_key = ('data_sharing_consent_needed', course_id)
if request.session.get(consent_key) is False:
return False
enterprise_learner_details = get_enterprise_learner_data(request.site, user)
if not enterprise_learner_details:
consent_needed = False
else:
client = ConsentApiClient()
consent_needed = any(
client.consent_required(
username=user.username,
course_id=course_id,
enterprise_customer_uuid=learner['enterprise_customer']['uuid'],
enrollment_exists=enrollment_exists,
)
for learner in enterprise_learner_details
)
if not consent_needed:
# Set an ephemeral item in the user's session to prevent us from needing
# to make a Consent API request every time this function is called.
request.session[consent_key] = False
return consent_needed
def get_enterprise_consent_url(request, course_id, user=None, return_to=None):
def get_enterprise_consent_url(request, course_id, user=None, return_to=None, enrollment_exists=False):
"""
Build a URL to redirect the user to the Enterprise app to provide data sharing
consent for a specific course ID.
......@@ -290,10 +385,13 @@ def get_enterprise_consent_url(request, course_id, user=None, return_to=None):
* return_to: url name label for the page to return to after consent is granted.
If None, return to request.path instead.
"""
if not enterprise_enabled():
return ''
if user is None:
user = request.user
if not consent_needed_for_course(user, course_id):
if not consent_needed_for_course(request, user, course_id, enrollment_exists=enrollment_exists):
return None
if return_to is None:
......@@ -318,30 +416,6 @@ def get_enterprise_consent_url(request, course_id, user=None, return_to=None):
return full_url
def get_cache_key(**kwargs):
"""
Get MD5 encoded cache key for given arguments.
Here is the format of key before MD5 encryption.
key1:value1__key2:value2 ...
Example:
>>> get_cache_key(site_domain="example.com", resource="enterprise-learner")
# Here is key format for above call
# "site_domain:example.com__resource:enterprise-learner"
a54349175618ff1659dee0978e3149ca
Arguments:
**kwargs: Key word arguments that need to be present in cache key.
Returns:
An MD5 encoded key uniquely identified by the key word arguments.
"""
key = '__'.join(['{}:{}'.format(item, value) for item, value in six.iteritems(kwargs)])
return hashlib.md5(key).hexdigest()
def get_enterprise_learner_data(site, user):
"""
Client API operation adapter/wrapper
......@@ -366,42 +440,40 @@ def get_dashboard_consent_notification(request, user, course_enrollments):
Returns:
str: Either an empty string, or a string containing the HTML code for the notification banner.
"""
if not enterprise_enabled():
return ''
enrollment = None
enterprise_enrollment = None
consent_needed = False
course_id = request.GET.get(CONSENT_FAILED_PARAMETER)
if course_id:
enterprise_customer = enterprise_customer_for_request(request)
if not enterprise_customer:
return ''
for course_enrollment in course_enrollments:
if str(course_enrollment.course_id) == course_id:
enrollment = course_enrollment
break
try:
enterprise_enrollment = EnterpriseCourseEnrollment.objects.get(
course_id=course_id,
enterprise_customer_user__user_id=user.id,
)
except EnterpriseCourseEnrollment.DoesNotExist:
pass
client = ConsentApiClient()
consent_needed = client.consent_required(
enterprise_customer_uuid=enterprise_customer['uuid'],
username=user.username,
course_id=course_id,
)
if enterprise_enrollment and enrollment:
enterprise_customer = enterprise_enrollment.enterprise_customer_user.enterprise_customer
contact_info = getattr(enterprise_customer, 'contact_email', None)
if consent_needed and enrollment:
if contact_info is None:
message_template = _(
'If you have concerns about sharing your data, please contact your administrator '
'at {enterprise_customer_name}.'
)
else:
message_template = _(
'If you have concerns about sharing your data, please contact your administrator '
'at {enterprise_customer_name} at {contact_info}.'
)
message_template = _(
'If you have concerns about sharing your data, please contact your administrator '
'at {enterprise_customer_name}.'
)
message = message_template.format(
enterprise_customer_name=enterprise_customer.name,
contact_info=contact_info,
enterprise_customer_name=enterprise_customer['name'],
)
title = _(
'Enrollment in {course_name} was not complete.'
......@@ -417,52 +489,3 @@ def get_dashboard_consent_notification(request, user, course_enrollments):
}
)
return ''
def is_course_in_enterprise_catalog(site, course_id, enterprise_catalog_id):
"""
Verify that the provided course id exists in the site base list of course
run keys from the provided enterprise course catalog.
Arguments:
course_id (str): The course ID.
site: (django.contrib.sites.Site) site instance
enterprise_catalog_id (Int): Course catalog id of enterprise
Returns:
Boolean
"""
cache_key = get_cache_key(
site_domain=site.domain,
resource='catalogs.contains',
course_id=course_id,
catalog_id=enterprise_catalog_id
)
response = cache.get(cache_key)
if not response:
catalog_integration = CatalogIntegration.current()
if not catalog_integration.enabled:
LOGGER.error("Catalog integration is not enabled.")
return False
try:
user = User.objects.get(username=catalog_integration.service_username)
except User.DoesNotExist:
LOGGER.exception("Catalog service user '%s' does not exist.", catalog_integration.service_username)
return False
try:
# GET: /api/v1/catalogs/{catalog_id}/contains?course_run_id={course_run_ids}
response = create_catalog_api_client(user=user).catalogs(enterprise_catalog_id).contains.get(
course_run_id=course_id
)
cache.set(cache_key, response, settings.COURSES_API_CACHE_TIMEOUT)
except (ConnectionError, SlumberBaseException, Timeout):
LOGGER.exception('Unable to connect to Course Catalog service for catalog contains endpoint.')
return False
try:
return response['courses'][course_id]
except KeyError:
return False
openedx/features/enterprise_support/tests/mixins/enterprise.py
......@@ -7,6 +7,8 @@ import mock
import httpretty
from django.conf import settings
from django.core.cache import cache
from django.core.urlresolvers import reverse
from django.test import SimpleTestCase
class EnterpriseServiceMockMixin(object):
......@@ -14,6 +16,8 @@ class EnterpriseServiceMockMixin(object):
Mocks for the Enterprise service responses.
"""
consent_url = '{}{}'.format(settings.ENTERPRISE_CONSENT_API_URL, 'data_sharing_consent')
def setUp(self):
super(EnterpriseServiceMockMixin, self).setUp()
cache.clear()
......@@ -23,6 +27,19 @@ class EnterpriseServiceMockMixin(object):
"""Return a URL to the configured Enterprise API. """
return '{}{}/'.format(settings.ENTERPRISE_API_URL, path)
def mock_get_enterprise_customer(self, uuid, response, status):
"""
Helper to mock the HTTP call to the /enterprise-customer/uuid endpoint
"""
body = json.dumps(response)
httpretty.register_uri(
method=httpretty.GET,
uri=(self.get_enterprise_url('enterprise-customer') + uuid + '/'),
body=body,
content_type='application/json',
status=status,
)
def mock_enterprise_course_enrollment_post_api( # pylint: disable=invalid-name
self,
username='test_user',
......@@ -57,6 +74,70 @@ class EnterpriseServiceMockMixin(object):
status=500
)
def mock_consent_response(
self,
username,
course_id,
ec_uuid,
method=httpretty.GET,
granted=True,
required=False,
exists=True,
response_code=None
):
response_body = {
'username': username,
'course_id': course_id,
'enterprise_customer_uuid': ec_uuid,
'consent_provided': granted,
'consent_required': required,
'exists': exists,
}
httpretty.register_uri(
method=method,
uri=self.consent_url,
content_type='application/json',
body=json.dumps(response_body),
status=response_code or 200,
)
def mock_consent_post(self, username, course_id, ec_uuid):
self.mock_consent_response(
username,
course_id,
ec_uuid,
method=httpretty.POST,
granted=True,
exists=True,
)
def mock_consent_get(self, username, course_id, ec_uuid):
self.mock_consent_response(
username,
course_id,
ec_uuid
)
def mock_consent_missing(self, username, course_id, ec_uuid):
self.mock_consent_response(
username,
course_id,
ec_uuid,
exists=False,
granted=False,
required=True,
)
def mock_consent_not_required(self, username, course_id, ec_uuid):
self.mock_consent_response(
username,
course_id,
ec_uuid,
exists=False,
granted=False,
required=False,
)
def mock_enterprise_learner_api(
self,
catalog_id=1,
......@@ -134,7 +215,7 @@ class EnterpriseServiceMockMixin(object):
)
class EnterpriseTestConsentRequired(object):
class EnterpriseTestConsentRequired(SimpleTestCase):
"""
Mixin to help test the data_sharing_consent_required decorator.
"""
......@@ -149,20 +230,30 @@ class EnterpriseTestConsentRequired(object):
* url: URL to test
* status_code: expected status code of URL when no data sharing consent is required.
"""
with mock.patch('openedx.features.enterprise_support.api.enterprise_enabled', return_value=True):
with mock.patch('openedx.features.enterprise_support.api.consent_necessary_for_course') as mock_consent_necessary: # pylint: disable=line-too-long
# Ensure that when consent is necessary, the user is redirected to the consent page.
mock_consent_necessary.return_value = True
response = client.get(url)
assert response.status_code == 302
assert 'grant_data_sharing_permissions' in response.url # pylint: disable=no-member
# Ensure that when consent is not necessary, the user continues through to the requested page.
mock_consent_necessary.return_value = False
response = client.get(url)
assert response.status_code == status_code
# If we were expecting a redirect, ensure it's not to the data sharing permission page
if status_code == 302:
assert 'grant_data_sharing_permissions' not in response.url # pylint: disable=no-member
return response
def mock_consent_reverse(*args, **kwargs):
if args[0] == 'grant_data_sharing_permissions':
return '/enterprise/grant_data_sharing_permissions'
return reverse(*args, **kwargs)
with mock.patch('openedx.features.enterprise_support.api.reverse', side_effect=mock_consent_reverse):
with mock.patch('openedx.features.enterprise_support.api.enterprise_enabled', return_value=True):
with mock.patch(
'openedx.features.enterprise_support.api.consent_needed_for_course'
) as mock_consent_necessary:
# Ensure that when consent is necessary, the user is redirected to the consent page.
mock_consent_necessary.return_value = True
response = client.get(url)
while(response.status_code == 302 and 'grant_data_sharing_permissions' not in response.url):
response = client.get(response.url)
self.assertEqual(response.status_code, 302)
self.assertIn('grant_data_sharing_permissions', response.url) # pylint: disable=no-member
# Ensure that when consent is not necessary, the user continues through to the requested page.
mock_consent_necessary.return_value = False
response = client.get(url)
self.assertEqual(response.status_code, status_code)
# If we were expecting a redirect, ensure it's not to the data sharing permission page
if status_code == 302:
self.assertNotIn('grant_data_sharing_permissions', response.url) # pylint: disable=no-member
return response
openedx/features/enterprise_support/tests/test_api.py
"""
Test the enterprise support APIs.
"""
import json
import unittest
import ddt
import httpretty
import mock
from django.conf import settings
from django.core.urlresolvers import reverse
from django.http import HttpResponseRedirect
from django.test import SimpleTestCase
from django.test.utils import override_settings
from openedx.features.enterprise_support.api import (
consent_needed_for_course,
data_sharing_consent_required,
enterprise_customer_for_request,
enterprise_enabled,
......@@ -16,80 +22,105 @@ from openedx.features.enterprise_support.api import (
get_enterprise_consent_url,
)
from openedx.features.enterprise_support.tests.mixins.enterprise import EnterpriseServiceMockMixin
from student.tests.factories import UserFactory
@ddt.ddt
@override_settings(ENABLE_ENTERPRISE_INTEGRATION=True)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
class TestEnterpriseApi(unittest.TestCase):
class TestEnterpriseApi(EnterpriseServiceMockMixin, SimpleTestCase):
"""
Test enterprise support APIs.
"""
@override_settings(ENABLE_ENTERPRISE_INTEGRATION=True)
@mock.patch('openedx.features.enterprise_support.api.EnterpriseCustomer')
@mock.patch('openedx.features.enterprise_support.api.get_partial_pipeline')
def test_enterprise_customer_for_request(self, pipeline_mock, ec_class_mock):
"""
Test that the correct EnterpriseCustomer, if any, is returned.
"""
def get_ec_mock(**kwargs):
by_provider_id_kw = 'enterprise_customer_identity_provider__provider_id'
provider_id = kwargs.get(by_provider_id_kw, '')
uuid = kwargs.get('uuid', '')
if uuid == 'real-uuid' or provider_id == 'real-provider-id':
return 'this-is-actually-an-enterprise-customer'
elif uuid == 'not-a-uuid':
raise ValueError
else:
raise Exception
ec_class_mock.DoesNotExist = Exception
ec_class_mock.objects.get.side_effect = get_ec_mock
pipeline_mock.return_value = None
request = mock.MagicMock()
request.GET.get.return_value = 'real-uuid'
self.assertEqual(enterprise_customer_for_request(request), 'this-is-actually-an-enterprise-customer')
request.GET.get.return_value = 'not-a-uuid'
self.assertEqual(enterprise_customer_for_request(request), None)
request.GET.get.return_value = 'fake-uuid'
self.assertEqual(enterprise_customer_for_request(request), None)
request.GET.get.return_value = None
self.assertEqual(
enterprise_customer_for_request(request, tpa_hint='real-provider-id'),
'this-is-actually-an-enterprise-customer'
def setUp(self):
UserFactory.create(
username='enterprise_worker',
email='ent_worker@example.com',
password='password123',
)
self.assertEqual(enterprise_customer_for_request(request, tpa_hint='fake-provider-id'), None)
self.assertEqual(enterprise_customer_for_request(request, tpa_hint=None), None)
@override_settings(ENABLE_ENTERPRISE_INTEGRATION=True)
super(TestEnterpriseApi, self).setUp()
@httpretty.activate
@override_settings(ENTERPRISE_SERVICE_WORKER_USERNAME='enterprise_worker')
def test_consent_needed_for_course(self):
user = mock.MagicMock(
username='janedoe',
is_authenticated=lambda: True,
)
request = mock.MagicMock(session={})
self.mock_enterprise_learner_api()
self.mock_consent_missing(user.username, 'fake-course', 'cf246b88-d5f6-4908-a522-fc307e0b0c59')
self.assertTrue(consent_needed_for_course(request, user, 'fake-course'))
self.mock_consent_get(user.username, 'fake-course', 'cf246b88-d5f6-4908-a522-fc307e0b0c59')
self.assertFalse(consent_needed_for_course(request, user, 'fake-course'))
# Test that the result is cached when false (remove the HTTP mock so if the result
# isn't cached, we'll fail spectacularly.)
httpretty.reset()
self.assertFalse(consent_needed_for_course(request, user, 'fake-course'))
@httpretty.activate
@mock.patch('openedx.features.enterprise_support.api.get_enterprise_learner_data')
@mock.patch('openedx.features.enterprise_support.api.EnterpriseCustomer')
@mock.patch('openedx.features.enterprise_support.api.Registry')
@mock.patch('openedx.features.enterprise_support.api.get_partial_pipeline')
def test_get_enterprise_customer_for_request_from_pipeline(self, pipeline_mock, registry_mock, ec_class_mock):
"""
Test that the correct EnterpriseCustomer, if any, is returned when
the user is in the middle of a third-party auth pipeline.
"""
def get_ec_mock(**kwargs):
by_provider_id_kw = 'enterprise_customer_identity_provider__provider_id'
provider_id = kwargs.get(by_provider_id_kw, '')
uuid = kwargs.get('uuid', '')
if uuid == 'real-uuid' or provider_id == 'real-provider-id':
# Only return the good value if we get the parameter we expect.
return 'this-is-actually-an-enterprise-customer'
@mock.patch('openedx.features.enterprise_support.api.Registry')
@override_settings(ENTERPRISE_SERVICE_WORKER_USERNAME='enterprise_worker')
def test_enterprise_customer_for_request(
self,
mock_registry,
mock_partial,
mock_ec_model,
mock_get_el_data
):
def mock_get_ec(**kwargs):
uuid = kwargs.get('enterprise_customer_identity_provider__provider_id')
if uuid:
return mock.MagicMock(uuid=uuid)
raise Exception
mock_ec_model.objects.get.side_effect = mock_get_ec
mock_ec_model.DoesNotExist = Exception
mock_partial.return_value = True
mock_registry.get_from_pipeline.return_value.provider_id = 'real-ent-uuid'
self.mock_get_enterprise_customer('real-ent-uuid', {"real": "enterprisecustomer"}, 200)
ec = enterprise_customer_for_request(mock.MagicMock())
self.assertEqual(ec, {"real": "enterprisecustomer"})
ec_class_mock.DoesNotExist = Exception
ec_class_mock.objects.get.side_effect = get_ec_mock
httpretty.reset()
# Truthy value from the pipeline getter to imitate a running pipeline
pipeline_mock.return_value = {"fake_pipeline": "sofake"}
self.mock_get_enterprise_customer('real-ent-uuid', {"detail": "Not found."}, 404)
provider_mock = registry_mock.get_from_pipeline.return_value
provider_mock.provider_id = 'real-provider-id'
ec = enterprise_customer_for_request(mock.MagicMock())
request = mock.MagicMock()
self.assertIsNone(ec)
self.assertEqual(enterprise_customer_for_request(request), 'this-is-actually-an-enterprise-customer')
mock_registry.get_from_pipeline.return_value.provider_id = None
httpretty.reset()
self.mock_get_enterprise_customer('real-ent-uuid', {"real": "enterprisecustomer"}, 200)
ec = enterprise_customer_for_request(mock.MagicMock(GET={"enterprise_customer": 'real-ent-uuid'}))
self.assertEqual(ec, {"real": "enterprisecustomer"})
ec = enterprise_customer_for_request(
mock.MagicMock(GET={}, COOKIES={settings.ENTERPRISE_CUSTOMER_COOKIE_NAME: 'real-ent-uuid'})
)
self.assertEqual(ec, {"real": "enterprisecustomer"})
mock_get_el_data.return_value = [{'enterprise_customer': {'uuid': 'real-ent-uuid'}}]
ec = enterprise_customer_for_request(
mock.MagicMock(GET={}, COOKIES={}, user=mock.MagicMock(is_authenticated=lambda: True), site=1)
)
self.assertEqual(ec, {"real": "enterprisecustomer"})
def check_data_sharing_consent(self, consent_required=False, consent_url=None):
"""
......@@ -120,7 +151,7 @@ class TestEnterpriseApi(unittest.TestCase):
self.assertEqual(response, (args, kwargs))
@mock.patch('openedx.features.enterprise_support.api.enterprise_enabled')
@mock.patch('openedx.features.enterprise_support.api.consent_necessary_for_course')
@mock.patch('openedx.features.enterprise_support.api.consent_needed_for_course')
def test_data_consent_required_enterprise_disabled(self,
mock_consent_necessary,
mock_enterprise_enabled):
......@@ -136,7 +167,7 @@ class TestEnterpriseApi(unittest.TestCase):
mock_consent_necessary.assert_not_called()
@mock.patch('openedx.features.enterprise_support.api.enterprise_enabled')
@mock.patch('openedx.features.enterprise_support.api.consent_necessary_for_course')
@mock.patch('openedx.features.enterprise_support.api.consent_needed_for_course')
def test_no_course_data_consent_required(self,
mock_consent_necessary,
mock_enterprise_enabled):
......@@ -154,7 +185,7 @@ class TestEnterpriseApi(unittest.TestCase):
mock_consent_necessary.assert_called_once()
@mock.patch('openedx.features.enterprise_support.api.enterprise_enabled')
@mock.patch('openedx.features.enterprise_support.api.consent_necessary_for_course')
@mock.patch('openedx.features.enterprise_support.api.consent_needed_for_course')
@mock.patch('openedx.features.enterprise_support.api.get_enterprise_consent_url')
def test_data_consent_required(self, mock_get_consent_url, mock_consent_necessary, mock_enterprise_enabled):
"""
......@@ -172,11 +203,18 @@ class TestEnterpriseApi(unittest.TestCase):
mock_enterprise_enabled.assert_called_once()
mock_consent_necessary.assert_called_once()
@mock.patch('openedx.features.enterprise_support.api.reverse')
@mock.patch('openedx.features.enterprise_support.api.consent_needed_for_course')
def test_get_enterprise_consent_url(self, needed_for_course_mock):
def test_get_enterprise_consent_url(self, needed_for_course_mock, reverse_mock):
"""
Verify that get_enterprise_consent_url correctly builds URLs.
"""
def fake_reverse(*args, **kwargs):
if args[0] == 'grant_data_sharing_permissions':
return '/enterprise/grant_data_sharing_permissions'
return reverse(*args, **kwargs)
reverse_mock.side_effect = fake_reverse
needed_for_course_mock.return_value = True
request_mock = mock.MagicMock(
......@@ -196,119 +234,60 @@ class TestEnterpriseApi(unittest.TestCase):
actual_url = get_enterprise_consent_url(request_mock, course_id, return_to=return_to)
self.assertEqual(actual_url, expected_url)
def test_get_dashboard_consent_notification_no_param(self):
"""
Test that the output of the consent notification renderer meets expectations.
"""
@ddt.data(
(False, {'real': 'enterprise', 'uuid': ''}, 'course', [], []),
(True, {}, 'course', [], []),
(True, {'real': 'enterprise'}, None, [], []),
(True, {'name': 'GriffCo', 'uuid': ''}, 'real-course', [], []),
(True, {'name': 'GriffCo', 'uuid': ''}, 'real-course', [mock.MagicMock(course_id='other-id')], []),
(
True,
{'name': 'GriffCo', 'uuid': 'real-uuid'},
'real-course',
[
mock.MagicMock(
course_id='real-course',
course_overview=mock.MagicMock(
display_name='My Cool Course'
)
)
],
[
'If you have concerns about sharing your data, please contact your administrator at GriffCo.',
'Enrollment in My Cool Course was not complete.'
]
),
)
@ddt.unpack
@mock.patch('openedx.features.enterprise_support.api.ConsentApiClient')
@mock.patch('openedx.features.enterprise_support.api.enterprise_customer_for_request')
def test_get_dashboard_consent_notification(
self,
consent_return_value,
enterprise_customer,
course_id,
enrollments,
expected_substrings,
ec_for_request,
consent_client_class
):
request = mock.MagicMock(
GET={}
)
notification_string = get_dashboard_consent_notification(
request, None, None
GET={'consent_failed': course_id}
)
self.assertEqual(notification_string, '')
consent_client = consent_client_class.return_value
consent_client.consent_required.return_value = consent_return_value
def test_get_dashboard_consent_notification_no_enrollments(self):
request = mock.MagicMock(
GET={'consent_failed': 'course-v1:edX+DemoX+Demo_Course'}
)
enrollments = []
user = mock.MagicMock(id=1)
notification_string = get_dashboard_consent_notification(
request, user, enrollments,
)
self.assertEqual(notification_string, '')
ec_for_request.return_value = enterprise_customer
def test_get_dashboard_consent_notification_no_matching_enrollments(self):
request = mock.MagicMock(
GET={'consent_failed': 'course-v1:edX+DemoX+Demo_Course'}
)
enrollments = [mock.MagicMock(course_id='other_course_id')]
user = mock.MagicMock(id=1)
notification_string = get_dashboard_consent_notification(
request, user, enrollments,
)
self.assertEqual(notification_string, '')
user = mock.MagicMock()
def test_get_dashboard_consent_notification_no_matching_ece(self):
request = mock.MagicMock(
GET={'consent_failed': 'course-v1:edX+DemoX+Demo_Course'}
)
enrollments = [mock.MagicMock(course_id='course-v1:edX+DemoX+Demo_Course')]
user = mock.MagicMock(id=1)
notification_string = get_dashboard_consent_notification(
request, user, enrollments,
)
self.assertEqual(notification_string, '')
@mock.patch('openedx.features.enterprise_support.api.EnterpriseCourseEnrollment')
def test_get_dashboard_consent_notification_no_contact_info(self, ece_mock):
mock_get_ece = ece_mock.objects.get
ece_mock.DoesNotExist = Exception
mock_ece = mock_get_ece.return_value
mock_ece.enterprise_customer_user = mock.MagicMock(
enterprise_customer=mock.MagicMock(
contact_email=None
)
)
mock_ec = mock_ece.enterprise_customer_user.enterprise_customer
mock_ec.name = 'Veridian Dynamics'
request = mock.MagicMock(
GET={'consent_failed': 'course-v1:edX+DemoX+Demo_Course'}
)
enrollments = [
mock.MagicMock(
course_id='course-v1:edX+DemoX+Demo_Course',
course_overview=mock.MagicMock(
display_name='edX Demo Course',
)
),
]
user = mock.MagicMock(id=1)
notification_string = get_dashboard_consent_notification(
request, user, enrollments,
)
expected_message = (
'If you have concerns about sharing your data, please contact your '
'administrator at Veridian Dynamics.'
)
self.assertIn(expected_message, notification_string)
expected_header = 'Enrollment in edX Demo Course was not complete.'
self.assertIn(expected_header, notification_string)
@mock.patch('openedx.features.enterprise_support.api.EnterpriseCourseEnrollment')
def test_get_dashboard_consent_notification_contact_info(self, ece_mock):
mock_get_ece = ece_mock.objects.get
ece_mock.DoesNotExist = Exception
mock_ece = mock_get_ece.return_value
mock_ece.enterprise_customer_user = mock.MagicMock(
enterprise_customer=mock.MagicMock(
contact_email='v.palmer@veridiandynamics.com'
)
)
mock_ec = mock_ece.enterprise_customer_user.enterprise_customer
mock_ec.name = 'Veridian Dynamics'
request = mock.MagicMock(
GET={'consent_failed': 'course-v1:edX+DemoX+Demo_Course'}
)
enrollments = [
mock.MagicMock(
course_id='course-v1:edX+DemoX+Demo_Course',
course_overview=mock.MagicMock(
display_name='edX Demo Course',
)
),
]
user = mock.MagicMock(id=1)
notification_string = get_dashboard_consent_notification(
request, user, enrollments,
)
expected_message = (
'If you have concerns about sharing your data, please contact your '
'administrator at Veridian Dynamics at v.palmer@veridiandynamics.com.'
)
self.assertIn(expected_message, notification_string)
expected_header = 'Enrollment in edX Demo Course was not complete.'
self.assertIn(expected_header, notification_string)
if expected_substrings:
for substr in expected_substrings:
self.assertIn(substr, notification_string)
else:
self.assertEqual(notification_string, '')
requirements/edx/base.txt
......@@ -48,7 +48,7 @@ edx-lint==0.4.3
astroid==1.3.8
edx-django-oauth2-provider==1.1.4
edx-django-sites-extensions==2.3.0
edx-enterprise==0.40.1
edx-enterprise==0.40.2
edx-oauth2-provider==1.2.0
edx-opaque-keys==0.4.0
edx-organizations==0.4.5
......
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