Files
edx-platform/openedx/features/enterprise_support/tests/test_api.py
Awais Qureshi 5ea60327b1 BOM-2336
Apply pylint-amnesty.
2021-02-04 16:49:47 +05:00

1077 lines
48 KiB
Python

"""
Test the enterprise support APIs.
"""
import ddt
import httpretty
import mock
from consent.models import DataSharingConsent
from django.conf import settings
from django.contrib.auth.models import User # lint-amnesty, pylint: disable=imported-auth-user
from django.core.cache import cache
from django.http import HttpResponseRedirect
from django.test.utils import override_settings
from django.urls import reverse
from edx_django_utils.cache import get_cache_key
from six.moves.urllib.parse import parse_qs
from slumber.exceptions import HttpClientError
from openedx.core.djangoapps.site_configuration.tests.factories import SiteFactory
from openedx.core.djangolib.testing.utils import CacheIsolationTestCase, skip_unless_lms
from openedx.features.enterprise_support.api import (
_CACHE_MISS,
ENTERPRISE_CUSTOMER_KEY_NAME,
EnterpriseApiException,
add_enterprise_customer_to_session,
ConsentApiClient,
ConsentApiServiceClient,
EnterpriseApiClient,
EnterpriseApiServiceClient,
consent_needed_for_course,
data_sharing_consent_required,
enterprise_customer_for_request,
enterprise_customer_from_api,
enterprise_customer_uuid_for_request,
enterprise_enabled,
get_consent_notification_data,
get_consent_required_courses,
get_dashboard_consent_notification,
get_enterprise_consent_url,
get_enterprise_learner_data_from_api,
get_enterprise_learner_data_from_db,
get_enterprise_learner_portal_enabled_message,
insert_enterprise_pipeline_elements,
unlink_enterprise_user_from_idp,
)
from openedx.features.enterprise_support.tests import FEATURES_WITH_ENTERPRISE_ENABLED
from openedx.features.enterprise_support.tests.factories import (
EnterpriseCustomerIdentityProviderFactory,
EnterpriseCustomerUserFactory,
)
from openedx.features.enterprise_support.tests.mixins.enterprise import EnterpriseServiceMockMixin
from openedx.features.enterprise_support.utils import clear_data_consent_share_cache
from common.djangoapps.student.tests.factories import UserFactory
from enterprise.models import EnterpriseCustomerUser # lint-amnesty, pylint: disable=wrong-import-order
class MockEnrollment(mock.MagicMock):
"""
Mock object for an enrollment which has a consistent string representation
suitable for use in ddt parameters.
"""
def __repr__(self):
return '<MockEnrollment course_id={}>'.format(getattr(self, 'course_id', None))
@ddt.ddt
@override_settings(FEATURES=FEATURES_WITH_ENTERPRISE_ENABLED)
@skip_unless_lms
class TestEnterpriseApi(EnterpriseServiceMockMixin, CacheIsolationTestCase):
"""
Test enterprise support APIs.
"""
ENABLED_CACHES = ['default']
@classmethod
def setUpTestData(cls):
cls.user = UserFactory.create(
username=settings.ENTERPRISE_SERVICE_WORKER_USERNAME,
email='ent_worker@example.com',
password='password123',
)
super(TestEnterpriseApi, cls).setUpTestData()
def _assert_api_service_client(self, api_client, mocked_jwt_builder):
"""
Verify that the provided api client uses the enterprise service user to generate
JWT token for auth.
"""
mocked_jwt_builder.return_value = 'test-token'
enterprise_service_user = User.objects.get(username=settings.ENTERPRISE_SERVICE_WORKER_USERNAME)
enterprise_api_service_client = api_client()
mocked_jwt_builder.assert_called_once_with(enterprise_service_user)
# pylint: disable=protected-access
self.assertEqual(enterprise_api_service_client.client._store['session'].auth.token, 'test-token')
def _assert_api_client_with_user(self, api_client, mocked_jwt_builder):
"""
Verify that the provided api client uses the expected user to generate
JWT token for auth.
"""
mocked_jwt_builder.return_value = 'test-token'
dummy_enterprise_user = UserFactory.create(
username='dummy-enterprise-user',
email='dummy-enterprise-user@example.com',
password='password123',
)
enterprise_api_service_client = api_client(dummy_enterprise_user)
mocked_jwt_builder.assert_called_once_with(dummy_enterprise_user)
# pylint: disable=protected-access
self.assertEqual(enterprise_api_service_client.client._store['session'].auth.token, 'test-token')
return enterprise_api_service_client
def _assert_get_enterprise_customer(self, api_client, enterprise_api_data_for_mock):
"""
DRY method to verify caching for get enterprise customer method.
"""
cache_key = get_cache_key(
resource='enterprise-customer',
resource_id=enterprise_api_data_for_mock['uuid'],
username=settings.ENTERPRISE_SERVICE_WORKER_USERNAME,
)
self.mock_get_enterprise_customer(enterprise_api_data_for_mock['uuid'], enterprise_api_data_for_mock, 200)
self._assert_get_enterprise_customer_with_cache(api_client, enterprise_api_data_for_mock, cache_key)
def _assert_get_enterprise_customer_with_cache(self, api_client, enterprise_customer_data, cache_key):
"""
DRY method to verify that get enterprise customer response is cached.
"""
cached_enterprise_customer = cache.get(cache_key)
self.assertIsNone(cached_enterprise_customer)
enterprise_customer = api_client.get_enterprise_customer(enterprise_customer_data['uuid'])
self.assertEqual(enterprise_customer_data, enterprise_customer)
cached_enterprise_customer = cache.get(cache_key)
self.assertEqual(cached_enterprise_customer, enterprise_customer)
@httpretty.activate
@mock.patch('openedx.features.enterprise_support.api.create_jwt_for_user')
def test_enterprise_api_client_with_service_user(self, mock_jwt_builder):
"""
Verify that enterprise API service client uses enterprcreate_jwt_for_userise service user
by default to authenticate and access enterprise API.
"""
self._assert_api_service_client(EnterpriseApiServiceClient, mock_jwt_builder)
# Verify that enterprise customer data is cached properly for the
# enterprise api client.
enterprise_api_client = EnterpriseApiServiceClient()
enterprise_api_data_for_mock_1 = {'name': 'dummy-enterprise-customer-1', 'uuid': 'enterprise-uuid-1'}
self._assert_get_enterprise_customer(enterprise_api_client, enterprise_api_data_for_mock_1)
# Now try to get enterprise customer for another enterprise and verify
# that enterprise api client returns data according to the provided
# enterprise UUID.
enterprise_api_data_for_mock_2 = {'name': 'dummy-enterprise-customer-2', 'uuid': 'enterprise-uuid-2'}
self._assert_get_enterprise_customer(enterprise_api_client, enterprise_api_data_for_mock_2)
@httpretty.activate
@mock.patch('openedx.features.enterprise_support.api.create_jwt_for_user')
def test_enterprise_api_client_with_user(self, mock_jwt_builder):
"""
Verify that enterprise API client uses the provided user to
authenticate and access enterprise API.
"""
self._assert_api_client_with_user(EnterpriseApiClient, mock_jwt_builder)
@ddt.data(True, False)
@httpretty.activate
@mock.patch('openedx.features.enterprise_support.api.create_jwt_for_user')
def test_enterprise_api_client_with_user_post_enrollment(self, should_raise_http_error, mock_jwt_builder):
"""
Verify that enterprise API client uses the provided user to
authenticate and access enterprise API.
"""
api_client = self._assert_api_client_with_user(EnterpriseApiClient, mock_jwt_builder)
setattr(api_client.client, 'enterprise-course-enrollment', mock.Mock())
mock_endpoint = getattr(api_client.client, 'enterprise-course-enrollment')
if should_raise_http_error:
mock_endpoint.post.side_effect = HttpClientError
username = 'spongebob'
course_id = 'burger-flipping-101'
consent_granted = True
if should_raise_http_error:
with self.assertRaises(EnterpriseApiException):
api_client.post_enterprise_course_enrollment(username, course_id, consent_granted)
else:
api_client.post_enterprise_course_enrollment(username, course_id, consent_granted)
mock_endpoint.post.assert_called_once_with(data={
'username': username,
'course_id': course_id,
'consent_granted': consent_granted,
})
@mock.patch('openedx.features.enterprise_support.api.enterprise_customer_uuid_for_request')
@mock.patch('openedx.features.enterprise_support.api.EnterpriseApiClient')
def test_enterprise_customer_from_api_cache_miss(self, mock_client_class, mock_uuid_from_request):
mock_uuid_from_request.return_value = _CACHE_MISS
mock_request = mock.Mock()
actual_result = enterprise_customer_from_api(mock_request)
self.assertIsNone(actual_result)
self.assertFalse(mock_client_class.called)
@httpretty.activate
@mock.patch('openedx.features.enterprise_support.api.create_jwt_for_user')
def test_enterprise_consent_api_client_with_service_user(self, mock_jwt_builder):
"""
Verify that enterprise API consent service client uses enterprise
service user by default to authenticate and access enterprise API.
"""
self._assert_api_service_client(ConsentApiServiceClient, mock_jwt_builder)
@httpretty.activate
@mock.patch('openedx.features.enterprise_support.api.create_jwt_for_user')
def test_enterprise_consent_api_client_with_user(self, mock_jwt_builder):
"""
Verify that enterprise API consent service client uses the provided
user to authenticate and access enterprise API.
"""
consent_client = self._assert_api_client_with_user(ConsentApiClient, mock_jwt_builder)
consent_client.consent_endpoint = mock.Mock()
kwargs = {
'foo': 'a',
'bar': 'b',
}
consent_client.provide_consent(**kwargs)
consent_client.revoke_consent(**kwargs)
consent_client.consent_endpoint.post.assert_called_once_with(kwargs)
consent_client.consent_endpoint.delete.assert_called_once_with(**kwargs)
@httpretty.activate
@mock.patch('openedx.features.enterprise_support.api.get_enterprise_learner_data_from_db')
def test_consent_needed_for_course(self, mock_get_enterprise_learner_data):
user = UserFactory(username='janedoe')
request = mock.MagicMock(
user=user,
site=SiteFactory(domain="example.com"),
session={},
COOKIES={},
GET={},
)
ec_uuid = 'cf246b88-d5f6-4908-a522-fc307e0b0c59'
course_id = 'fake-course'
mock_get_enterprise_learner_data.return_value = self.get_mock_enterprise_learner_results()
self.mock_enterprise_learner_api()
# test that consent is not required for a non-enterprise customer
self.mock_consent_not_required(user.username, course_id, ec_uuid)
self.assertFalse(consent_needed_for_course(request, user, course_id))
# test required and missing consent for example now he becomes a enterprise customer
self.mock_consent_missing(user.username, course_id, ec_uuid)
# still result should be False as it has been stored in cache "Not to show consent", so it will confirm that
# cache is working fine
self.assertFalse(consent_needed_for_course(request, user, course_id))
# Removing cache
clear_data_consent_share_cache(user.id, course_id)
# Now test again
self.assertTrue(consent_needed_for_course(request, user, course_id))
# test after consent permission is granted
self.mock_consent_get(user.username, course_id, ec_uuid)
self.assertFalse(consent_needed_for_course(request, user, course_id))
# test when the enrollment already exists without a consent record existing.
clear_data_consent_share_cache(user.id, course_id)
self.mock_consent_missing(user.username, course_id, ec_uuid)
self.assertFalse(consent_needed_for_course(request, user, course_id, enrollment_exists=True))
@httpretty.activate
@mock.patch('openedx.features.enterprise_support.api.get_enterprise_learner_data_from_db')
def test_consent_needed_for_course_no_learner_data(self, mock_get_enterprise_learner_data):
user = UserFactory(username='janedoe')
request = mock.MagicMock(
user=user,
site=SiteFactory(domain="example.com"),
session={},
COOKIES={},
GET={},
)
ec_uuid = 'cf246b88-d5f6-4908-a522-fc307e0b0c59'
course_id = 'fake-course'
mock_get_enterprise_learner_data.return_value = None
self.mock_enterprise_learner_api()
# test that consent is not required for a non-enterprise customer
self.mock_consent_not_required(user.username, course_id, ec_uuid)
self.assertFalse(consent_needed_for_course(request, user, course_id))
@httpretty.activate
@mock.patch('enterprise.models.EnterpriseCustomer.catalog_contains_course')
def test_get_consent_required_courses(self, mock_catalog_contains_course):
mock_catalog_contains_course.return_value = True
user = UserFactory()
enterprise_customer_user = EnterpriseCustomerUserFactory(user_id=user.id)
course_id = 'fake-course'
data_sharing_consent = DataSharingConsent(
course_id=course_id,
enterprise_customer=enterprise_customer_user.enterprise_customer,
username=user.username,
granted=False
)
data_sharing_consent.save()
consent_required = get_consent_required_courses(user, [course_id])
self.assertIn(course_id, consent_required)
# now grant consent and call our method again
data_sharing_consent.granted = True
data_sharing_consent.save()
consent_required = get_consent_required_courses(user, [course_id])
self.assertNotIn(course_id, consent_required)
def test_consent_not_required_for_non_enterprise_user(self):
user = UserFactory()
course_id = 'fake-course'
consent_required_courses = get_consent_required_courses(user, [course_id])
assert set() == consent_required_courses
@mock.patch('openedx.features.enterprise_support.api.create_jwt_for_user')
def test_fetch_enterprise_learner_data_unauthenticated(self, mock_jwt_builder):
api_client = self._assert_api_client_with_user(EnterpriseApiClient, mock_jwt_builder)
setattr(api_client.client, 'enterprise-learner', mock.Mock())
mock_endpoint = getattr(api_client.client, 'enterprise-learner')
user = mock.Mock(is_authenticated=False)
self.assertIsNone(api_client.fetch_enterprise_learner_data(user))
self.assertFalse(mock_endpoint.called)
@mock.patch('openedx.features.enterprise_support.api.create_jwt_for_user')
def test_fetch_enterprise_learner_data(self, mock_jwt_builder):
api_client = self._assert_api_client_with_user(EnterpriseApiClient, mock_jwt_builder)
setattr(api_client.client, 'enterprise-learner', mock.Mock())
mock_endpoint = getattr(api_client.client, 'enterprise-learner')
user = mock.Mock(is_authenticated=True, username='spongebob')
response = api_client.fetch_enterprise_learner_data(user)
assert mock_endpoint.return_value.get.return_value == response
mock_endpoint.return_value.get.assert_called_once_with(username=user.username)
@mock.patch('openedx.features.enterprise_support.api.get_current_request')
@mock.patch('openedx.features.enterprise_support.api.create_jwt_for_user')
def test_fetch_enterprise_learner_data_http_error(self, mock_jwt_builder, mock_get_current_request):
api_client = self._assert_api_client_with_user(EnterpriseApiClient, mock_jwt_builder)
setattr(api_client.client, 'enterprise-learner', mock.Mock())
mock_endpoint = getattr(api_client.client, 'enterprise-learner')
mock_endpoint.return_value.get.side_effect = HttpClientError
mock_get_current_request.return_value.META = {
'PATH_INFO': 'whatever',
}
user = mock.Mock(is_authenticated=True, username='spongebob')
self.assertIsNone(api_client.fetch_enterprise_learner_data(user))
mock_endpoint.return_value.get.assert_called_once_with(username=user.username)
@mock.patch('openedx.features.enterprise_support.api.EnterpriseApiClient')
def test_get_enterprise_learner_data_from_api(self, mock_api_client_class):
user = mock.Mock(is_authenticated=True)
mock_client = mock_api_client_class.return_value
mock_client.fetch_enterprise_learner_data.return_value = {
'results': 'the-learner-data',
}
learner_data = get_enterprise_learner_data_from_api(user)
assert 'the-learner-data' == learner_data
mock_api_client_class.assert_called_once_with(user=user)
mock_client.fetch_enterprise_learner_data.assert_called_once_with(user)
def test_get_enterprise_learner_data_from_db_no_data(self):
assert [] == get_enterprise_learner_data_from_db(self.user)
def test_get_enterprise_learner_data_from_db(self):
enterprise_customer_user = EnterpriseCustomerUserFactory(user_id=self.user.id) # lint-amnesty, pylint: disable=unused-variable
user_data = get_enterprise_learner_data_from_db(self.user)[0]['user']
assert user_data['username'] == self.user.username
@httpretty.activate
@mock.patch('openedx.features.enterprise_support.api.get_enterprise_learner_data_from_db')
@mock.patch('openedx.features.enterprise_support.api.EnterpriseCustomer')
@mock.patch('openedx.features.enterprise_support.api.get_partial_pipeline')
@mock.patch('openedx.features.enterprise_support.api.Registry')
def test_enterprise_customer_for_request(
self,
mock_registry,
mock_partial,
mock_enterprise_customer_model,
mock_get_enterprise_learner_data,
):
def mock_get_enterprise_customer(**kwargs):
uuid = kwargs.get('enterprise_customer_identity_providers__provider_id')
if uuid:
return mock.MagicMock(uuid=uuid, user=self.user)
raise Exception
dummy_request = mock.MagicMock(session={}, user=self.user)
mock_enterprise_customer_model.objects.get.side_effect = mock_get_enterprise_customer
mock_enterprise_customer_model.DoesNotExist = Exception
mock_partial.return_value = True
mock_registry.get_from_pipeline.return_value.provider_id = 'real-ent-uuid'
# Verify that the method `enterprise_customer_for_request` returns
# expected enterprise customer against the requesting user.
self.mock_get_enterprise_customer('real-ent-uuid', {'real': 'enterprisecustomer'}, 200)
enterprise_customer = enterprise_customer_for_request(dummy_request)
self.assertEqual(enterprise_customer, {'real': 'enterprisecustomer'})
httpretty.reset()
# Verify that the method `enterprise_customer_for_request` returns no
# enterprise customer if the enterprise customer API throws 404.
del dummy_request.session['enterprise_customer']
self.mock_get_enterprise_customer('real-ent-uuid', {'detail': 'Not found.'}, 404)
enterprise_customer = enterprise_customer_for_request(dummy_request)
self.assertIsNone(enterprise_customer)
httpretty.reset()
# Verify that the method `enterprise_customer_for_request` returns
# expected enterprise customer against the requesting user even if
# the third-party auth pipeline has no `provider_id`.
mock_registry.get_from_pipeline.return_value.provider_id = None
self.mock_get_enterprise_customer('real-ent-uuid', {'real': 'enterprisecustomer'}, 200)
mock_request = mock.MagicMock(
GET={'enterprise_customer': 'real-ent-uuid'},
COOKIES={},
session={},
user=self.user
)
enterprise_customer = enterprise_customer_for_request(mock_request)
self.assertEqual(enterprise_customer, {'real': 'enterprisecustomer'})
# Verify that the method `enterprise_customer_for_request` returns
# expected enterprise customer against the requesting user even if
# the third-party auth pipeline has no `provider_id` but there is
# enterprise customer UUID in the cookie.
mock_request = mock.MagicMock(
GET={},
COOKIES={settings.ENTERPRISE_CUSTOMER_COOKIE_NAME: 'real-ent-uuid'},
session={},
user=self.user
)
enterprise_customer = enterprise_customer_for_request(mock_request)
self.assertEqual(enterprise_customer, {'real': 'enterprisecustomer'})
# Verify that the method `enterprise_customer_for_request` returns
# expected enterprise customer against the requesting user if
# data is cached only in the request session
mock_registry.get_from_pipeline.return_value.provider_id = None
self.mock_get_enterprise_customer('real-ent-uuid', {'real': 'enterprisecustomer'}, 200)
mock_request = mock.MagicMock(
GET={},
COOKIES={},
session={'enterprise_customer': {'real': 'enterprisecustomer'}},
user=self.user
)
enterprise_customer = enterprise_customer_for_request(mock_request)
self.assertEqual(enterprise_customer, {'real': 'enterprisecustomer'})
# Verify that we can still get enterprise customer from enterprise
# learner API even if we are unable to get it from preferred sources,
# e.g. url query parameters, third-party auth pipeline, enterprise
# cookie, or session.
mock_get_enterprise_learner_data.return_value = [{'enterprise_customer': {'uuid': 'real-ent-uuid'}}]
mock_request = mock.MagicMock(
GET={},
COOKIES={},
session={},
user=self.user,
site=1
)
enterprise_customer = enterprise_customer_for_request(mock_request)
self.assertEqual(enterprise_customer, {'real': 'enterprisecustomer'})
def test_enterprise_customer_for_request_with_session(self):
"""
Verify enterprise_customer_for_request stores and retrieves data from session appropriately
"""
dummy_request = mock.MagicMock(session={}, user=self.user)
enterprise_data = {'name': 'dummy-enterprise-customer', 'uuid': '8dc65e66-27c9-447b-87ff-ede6d66e3a5d'}
# Verify enterprise customer data fetched from API when it is not available in session
with mock.patch(
'openedx.features.enterprise_support.api.enterprise_customer_from_api',
return_value=enterprise_data
):
self.assertEqual(dummy_request.session.get('enterprise_customer'), None)
enterprise_customer = enterprise_customer_for_request(dummy_request)
self.assertEqual(enterprise_customer, enterprise_data)
self.assertEqual(dummy_request.session.get('enterprise_customer'), enterprise_data)
# Verify enterprise customer data fetched from session for subsequent calls
with mock.patch(
'openedx.features.enterprise_support.api.enterprise_customer_from_api',
return_value=enterprise_data
) as mock_enterprise_customer_from_api, mock.patch(
'openedx.features.enterprise_support.api.enterprise_customer_from_session',
return_value=enterprise_data
) as mock_enterprise_customer_from_session:
enterprise_customer = enterprise_customer_for_request(dummy_request)
self.assertEqual(enterprise_customer, enterprise_data)
self.assertEqual(mock_enterprise_customer_from_api.called, False)
self.assertEqual(mock_enterprise_customer_from_session.called, True)
# Verify enterprise customer data fetched from session for subsequent calls
# with unauthenticated user in SAML case
del dummy_request.user
with mock.patch(
'openedx.features.enterprise_support.api.enterprise_customer_from_api',
return_value=enterprise_data
) as mock_enterprise_customer_from_api, mock.patch(
'openedx.features.enterprise_support.api.enterprise_customer_from_session',
return_value=enterprise_data
) as mock_enterprise_customer_from_session:
enterprise_customer = enterprise_customer_for_request(dummy_request)
self.assertEqual(enterprise_customer, enterprise_data)
self.assertEqual(mock_enterprise_customer_from_api.called, False)
self.assertEqual(mock_enterprise_customer_from_session.called, True)
def check_data_sharing_consent(self, consent_required=False, consent_url=None):
"""
Used to test the data_sharing_consent_required view decorator.
"""
# Test by wrapping a function that has the expected signature
@data_sharing_consent_required
def view_func(request, course_id, *args, **kwargs):
"""
Return the function arguments, so they can be tested.
"""
return ((request, course_id,) + args, kwargs)
# Call the wrapped function
args = (mock.MagicMock(), 'course-id', 'another arg', 'and another')
kwargs = dict(a=1, b=2, c=3)
response = view_func(*args, **kwargs)
# If consent required, then the response should be a redirect to the consent URL, and the view function would
# not be called.
if consent_required:
self.assertIsInstance(response, HttpResponseRedirect)
self.assertEqual(response.url, consent_url) # pylint: disable=no-member
# Otherwise, the view function should have been called with the expected arguments.
else:
self.assertEqual(response, (args, kwargs))
@mock.patch('openedx.features.enterprise_support.api.enterprise_enabled')
@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):
"""
Verify that the wrapped view is called directly when enterprise integration is disabled,
without checking for course consent necessary.
"""
mock_enterprise_enabled.return_value = False
self.check_data_sharing_consent(consent_required=False)
mock_enterprise_enabled.assert_called_once()
mock_consent_necessary.assert_not_called()
@mock.patch('openedx.features.enterprise_support.api.enterprise_enabled')
@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):
"""
Verify that the wrapped view is called directly when enterprise integration is enabled,
and no course consent is required.
"""
mock_enterprise_enabled.return_value = True
mock_consent_necessary.return_value = False
self.check_data_sharing_consent(consent_required=False)
mock_enterprise_enabled.assert_called_once()
mock_consent_necessary.assert_called_once()
@mock.patch('openedx.features.enterprise_support.api.enterprise_enabled')
@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):
"""
Verify that the wrapped function returns a redirect to the consent URL when enterprise integration is enabled,
and course consent is required.
"""
mock_enterprise_enabled.return_value = True
mock_consent_necessary.return_value = True
consent_url = '/abc/def'
mock_get_consent_url.return_value = consent_url
self.check_data_sharing_consent(consent_required=True, consent_url=consent_url)
mock_get_consent_url.assert_called_once()
@ddt.data(True, False)
@httpretty.activate
@mock.patch('openedx.features.enterprise_support.api.enterprise_customer_uuid_for_request')
@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,
is_return_to_null,
needed_for_course_mock,
reverse_mock,
enterprise_customer_uuid_for_request_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)
enterprise_customer_uuid_for_request_mock.return_value = 'cf246b88-d5f6-4908-a522-fc307e0b0c59'
reverse_mock.side_effect = fake_reverse
needed_for_course_mock.return_value = True
request_mock = mock.MagicMock(
user=self.user,
path='/request_path',
build_absolute_uri=lambda x: 'http://localhost:8000' + x # Don't do it like this in prod. Ever.
)
course_id = 'course-v1:edX+DemoX+Demo_Course'
return_to = None if is_return_to_null else 'info'
expected_path = request_mock.path if is_return_to_null else '/courses/course-v1:edX+DemoX+Demo_Course/info'
expected_url_args = {
'course_id': ['course-v1:edX+DemoX+Demo_Course'],
'failure_url': ['http://localhost:8000/dashboard?consent_failed=course-v1%3AedX%2BDemoX%2BDemo_Course'],
'enterprise_customer_uuid': ['cf246b88-d5f6-4908-a522-fc307e0b0c59'],
'next': ['http://localhost:8000{}'.format(expected_path)]
}
actual_url = get_enterprise_consent_url(request_mock, course_id, return_to=return_to)
actual_url_args = parse_qs(actual_url.split('/enterprise/grant_data_sharing_permissions?')[1])
self.assertEqual(actual_url_args, expected_url_args)
@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', [MockEnrollment(course_id='other-id')], [], "", ""),
(
True,
{'name': 'GriffCo', 'uuid': 'real-uuid'},
'real-course',
[
MockEnrollment(
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.'
],
"", ""
),
(
True,
{'name': 'GriffCo', 'uuid': 'real-uuid'},
'real-course',
[
MockEnrollment(
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.'
],
"Title from DataSharingConsentTextOverrides model in consent app",
"Message from DataSharingConsentTextOverrides model in consent app"
),
)
@ddt.unpack
@mock.patch('openedx.features.enterprise_support.api.ConsentApiClient')
@mock.patch('openedx.features.enterprise_support.api.enterprise_customer_for_request')
@mock.patch('openedx.features.enterprise_support.api.get_consent_notification_data')
def test_get_dashboard_consent_notification(
self,
consent_return_value,
enterprise_customer,
course_id,
enrollments,
expected_substrings,
notification_title,
notification_message,
consent_notification_data,
ec_for_request,
consent_client_class
):
request = mock.MagicMock(
GET={'consent_failed': course_id}
)
consent_notification_data.return_value = notification_title, notification_message
consent_client = consent_client_class.return_value
consent_client.consent_required.return_value = consent_return_value
ec_for_request.return_value = enterprise_customer
user = mock.MagicMock()
notification_string = get_dashboard_consent_notification(
request, user, enrollments,
)
if notification_message and notification_title:
self.assertIn(notification_title, notification_string)
self.assertIn(notification_message, notification_string)
elif expected_substrings:
for substr in expected_substrings:
self.assertIn(substr, notification_string)
else:
self.assertEqual(notification_string, '')
@override_settings(FEATURES=dict(ENABLE_ENTERPRISE_INTEGRATION=False))
def test_utils_with_enterprise_disabled(self):
"""
Test that disabling the enterprise integration flag causes
the utilities to return the expected default values.
"""
self.assertFalse(enterprise_enabled())
self.assertEqual(insert_enterprise_pipeline_elements(None), None)
def test_utils_with_enterprise_enabled(self):
"""
Test that enabling enterprise integration (which is currently on by default) causes the
the utilities to return the expected values.
"""
self.assertTrue(enterprise_enabled())
pipeline = ['abc', 'social_core.pipeline.social_auth.load_extra_data', 'def']
insert_enterprise_pipeline_elements(pipeline)
self.assertEqual(pipeline, ['abc',
'enterprise.tpa_pipeline.handle_enterprise_logistration',
'social_core.pipeline.social_auth.load_extra_data',
'def'])
@mock.patch('openedx.features.enterprise_support.api.get_enterprise_learner_data_from_db')
def test_enterprise_learner_portal_message_cache_miss_no_customer(self, mock_learner_data_from_db):
"""
When no customer data exists in the request session _and_
no customer is associated with the requesting user, then ``get_enterprise_learner_portal_enabled_message()``
should return None.
"""
mock_request = mock.Mock(session={})
mock_learner_data_from_db.return_value = None
actual_result = get_enterprise_learner_portal_enabled_message(mock_request)
self.assertIsNone(actual_result)
mock_learner_data_from_db.assert_called_once_with(mock_request.user)
@mock.patch('openedx.features.enterprise_support.api.get_enterprise_learner_data_from_db')
@override_settings(ENTERPRISE_LEARNER_PORTAL_BASE_URL='http://localhost')
def test_enterprise_learner_portal_message_cache_miss_customer_exists(self, mock_learner_data_from_db):
"""
When no customer data exists in the request session but a
customer is associated with the requesting user, then ``get_enterprise_learner_portal_enabled_message()``
should return an appropriate message for that customer.
"""
mock_request = mock.Mock(session={})
mock_enterprise_customer = {
'uuid': 'some-uuid',
'name': 'Best Corp',
'enable_learner_portal': True,
'slug': 'best-corp',
}
mock_learner_data_from_db.return_value = [
{
'enterprise_customer': mock_enterprise_customer,
},
]
actual_result = get_enterprise_learner_portal_enabled_message(mock_request)
self.assertIn('custom dashboard for learning', actual_result)
self.assertIn('Best Corp', actual_result)
mock_learner_data_from_db.assert_called_once_with(mock_request.user)
# assert we cached the enterprise customer data in the request session after fetching it
assert mock_request.session.get(ENTERPRISE_CUSTOMER_KEY_NAME) == mock_enterprise_customer
@mock.patch('openedx.features.enterprise_support.api.get_enterprise_learner_data_from_db')
def test_enterprise_learner_portal_message_cache_hit_no_customer(self, mock_learner_data_from_db):
"""
When customer data exists in the request session but it's null/empty,
then ``get_enterprise_learner_portal_enabled_message()`` should return None.
"""
mock_request = mock.Mock(session={
ENTERPRISE_CUSTOMER_KEY_NAME: None,
})
actual_result = get_enterprise_learner_portal_enabled_message(mock_request)
self.assertIsNone(actual_result)
self.assertFalse(mock_learner_data_from_db.called)
@ddt.data(True, False)
@mock.patch('openedx.features.enterprise_support.api.get_enterprise_learner_data_from_db')
@override_settings(ENTERPRISE_LEARNER_PORTAL_BASE_URL='http://localhost')
def test_enterprise_learner_portal_message_cache_hit_customer_exists(
self, enable_learner_portal, mock_learner_data_from_db
):
"""
When customer data exists in the request session and it's a non-empty customer,
then ``get_enterprise_learner_portal_enabled_message()`` should return
an appropriate message for that customer.
"""
mock_enterprise_customer = {
'uuid': 'some-uuid',
'name': 'Best Corp',
'enable_learner_portal': enable_learner_portal,
'slug': 'best-corp',
}
mock_request = mock.Mock(session={
ENTERPRISE_CUSTOMER_KEY_NAME: mock_enterprise_customer,
})
actual_result = get_enterprise_learner_portal_enabled_message(mock_request)
if not enable_learner_portal:
self.assertIsNone(actual_result)
else:
self.assertIn('custom dashboard for learning', actual_result)
self.assertIn('Best Corp', actual_result)
self.assertFalse(mock_learner_data_from_db.called)
@mock.patch('openedx.features.enterprise_support.api.get_partial_pipeline', return_value=None)
def test_customer_uuid_for_request_sso_provider_id_customer_exists(self, mock_partial_pipeline):
mock_idp = EnterpriseCustomerIdentityProviderFactory.create()
mock_customer = mock_idp.enterprise_customer
mock_request = mock.Mock(
GET={'tpa_hint': mock_idp.provider_id},
COOKIES={},
session={},
)
actual_uuid = enterprise_customer_uuid_for_request(mock_request)
expected_uuid = mock_customer.uuid
self.assertEqual(expected_uuid, actual_uuid)
mock_partial_pipeline.assert_called_once_with(mock_request)
self.assertNotIn(ENTERPRISE_CUSTOMER_KEY_NAME, mock_request.session)
@mock.patch('openedx.features.enterprise_support.api.get_enterprise_learner_data_from_db')
@mock.patch('openedx.features.enterprise_support.api.get_partial_pipeline', return_value=None)
def test_customer_uuid_for_request_sso_provider_id_customer_non_existent_but_exist_in_db(
self,
mock_partial_pipeline,
mock_data_from_db,
):
enterprise_customer_uuid = 'adab9a14-f263-42e6-a234-db707026c4a6'
mock_request = mock.Mock(
GET={'tpa_hint': 'my-third-party-auth'},
COOKIES={},
session={},
)
mock_data_from_db.return_value = [
{'enterprise_customer': {'uuid': enterprise_customer_uuid}},
]
actual_uuid = enterprise_customer_uuid_for_request(mock_request)
self.assertEqual(actual_uuid, enterprise_customer_uuid)
mock_partial_pipeline.assert_called_once_with(mock_request)
self.assertIn(ENTERPRISE_CUSTOMER_KEY_NAME, mock_request.session)
@mock.patch('openedx.features.enterprise_support.api.get_partial_pipeline', return_value=None)
def test_enterprise_uuid_for_request_from_query_params(self, mock_partial_pipeline):
expected_uuid = 'my-uuid'
mock_request = mock.Mock(
GET={ENTERPRISE_CUSTOMER_KEY_NAME: expected_uuid},
COOKIES={},
session={},
)
actual_uuid = enterprise_customer_uuid_for_request(mock_request)
self.assertEqual(expected_uuid, actual_uuid)
mock_partial_pipeline.assert_called_once_with(mock_request)
self.assertNotIn(ENTERPRISE_CUSTOMER_KEY_NAME, mock_request.session)
@mock.patch('openedx.features.enterprise_support.api.get_partial_pipeline', return_value=None)
def test_enterprise_uuid_for_request_from_cookies(self, mock_partial_pipeline):
expected_uuid = 'my-uuid'
mock_request = mock.Mock(
GET={},
COOKIES={settings.ENTERPRISE_CUSTOMER_COOKIE_NAME: expected_uuid},
session={},
)
actual_uuid = enterprise_customer_uuid_for_request(mock_request)
self.assertEqual(expected_uuid, actual_uuid)
mock_partial_pipeline.assert_called_once_with(mock_request)
self.assertNotIn(ENTERPRISE_CUSTOMER_KEY_NAME, mock_request.session)
@mock.patch('openedx.features.enterprise_support.api.get_partial_pipeline', return_value=None)
def test_enterprise_uuid_for_request_from_session(self, mock_partial_pipeline):
expected_uuid = 'my-uuid'
mock_request = mock.Mock(
GET={},
COOKIES={},
session={ENTERPRISE_CUSTOMER_KEY_NAME: {'uuid': expected_uuid}},
)
actual_uuid = enterprise_customer_uuid_for_request(mock_request)
self.assertEqual(expected_uuid, actual_uuid)
mock_partial_pipeline.assert_called_once_with(mock_request)
self.assertEqual({'uuid': expected_uuid}, mock_request.session.get(ENTERPRISE_CUSTOMER_KEY_NAME))
@mock.patch('openedx.features.enterprise_support.api.get_enterprise_learner_data_from_db')
@mock.patch('openedx.features.enterprise_support.api.get_partial_pipeline', return_value=None)
def test_enterprise_uuid_for_request_cache_miss_but_exists_in_db(self, mock_partial_pipeline, mock_data_from_db):
mock_request = mock.Mock(
GET={},
COOKIES={},
session={},
)
mock_data_from_db.return_value = [
{'enterprise_customer': {'uuid': 'my-uuid'}},
]
actual_uuid = enterprise_customer_uuid_for_request(mock_request)
expected_uuid = 'my-uuid'
self.assertEqual(expected_uuid, actual_uuid)
mock_partial_pipeline.assert_called_once_with(mock_request)
mock_data_from_db.assert_called_once_with(mock_request.user)
self.assertEqual({'uuid': 'my-uuid'}, mock_request.session[ENTERPRISE_CUSTOMER_KEY_NAME])
@ddt.data(True, False)
@mock.patch('openedx.features.enterprise_support.api.get_enterprise_learner_data_from_db', return_value=None)
@mock.patch('openedx.features.enterprise_support.api.get_partial_pipeline', return_value=None)
def test_enterprise_uuid_for_request_cache_miss_non_existent(
self,
is_user_authenticated,
mock_partial_pipeline,
mock_data_from_db
):
mock_request = mock.Mock(
GET={},
COOKIES={},
session={},
)
mock_request.user.is_authenticated = is_user_authenticated
actual_uuid = enterprise_customer_uuid_for_request(mock_request)
self.assertIsNone(actual_uuid)
mock_partial_pipeline.assert_called_once_with(mock_request)
if is_user_authenticated:
mock_data_from_db.assert_called_once_with(mock_request.user)
self.assertIsNone(mock_request.session[ENTERPRISE_CUSTOMER_KEY_NAME])
else:
self.assertFalse(mock_data_from_db.called)
self.assertNotIn(ENTERPRISE_CUSTOMER_KEY_NAME, mock_request.session)
def test_enterprise_customer_from_session(self):
mock_request = mock.Mock(
GET={},
COOKIES={},
session={},
)
mock_request.user.is_authenticated = True
enterprise_customer = {
'name': 'abc',
'uuid': 'cf246b88-d5f6-4908-a522-fc307e0b0c59'
}
# set enterprise customer info with authenticate user
add_enterprise_customer_to_session(mock_request, enterprise_customer)
self.assertEqual(mock_request.session[ENTERPRISE_CUSTOMER_KEY_NAME], enterprise_customer)
# Now try to set info with un-authenticated user
mock_request.user.is_authenticated = False
add_enterprise_customer_to_session(mock_request, None)
# verify that existing session value should not be updated for un-authenticate user
self.assertEqual(mock_request.session[ENTERPRISE_CUSTOMER_KEY_NAME], enterprise_customer)
def test_get_consent_notification_data_no_overrides(self):
enterprise_customer = {
'name': 'abc',
'uuid': 'cf246b88-d5f6-4908-a522-fc307e0b0c59'
}
title_template, message_template = get_consent_notification_data(enterprise_customer)
self.assertIsNone(title_template)
self.assertIsNone(message_template)
@mock.patch('openedx.features.enterprise_support.api.DataSharingConsentTextOverrides')
def test_get_consent_notification_data(self, mock_override_model):
enterprise_customer = {
'name': 'abc',
'uuid': 'cf246b88-d5f6-4908-a522-fc307e0b0c59'
}
mock_override = mock.Mock(
declined_notification_title='the title',
declined_notification_message='the message',
)
mock_override_model.objects.get.return_value = mock_override
title_template, message_template = get_consent_notification_data(enterprise_customer)
assert mock_override.declined_notification_title == title_template
assert mock_override.declined_notification_message == message_template
@mock.patch('openedx.features.enterprise_support.api.Registry')
@mock.patch('openedx.features.enterprise_support.api.enterprise_customer_for_request')
def test_unlink_enterprise_user_from_idp(self, mock_customer_from_request, mock_registry):
customer_idp = EnterpriseCustomerIdentityProviderFactory.create(
provider_id='the-provider',
)
customer = customer_idp.enterprise_customer
customer_user = EnterpriseCustomerUserFactory.create( # lint-amnesty, pylint: disable=unused-variable
enterprise_customer=customer,
user_id=self.user.id,
)
mock_customer_from_request.return_value = {
'uuid': customer.uuid,
}
mock_registry.get_enabled_by_backend_name.return_value = [
mock.Mock(provider_id='the-provider')
]
request = mock.Mock()
unlink_enterprise_user_from_idp(request, self.user, idp_backend_name='the-backend-name')
assert 0 == EnterpriseCustomerUser.objects.filter(user_id=self.user.id).count()
@mock.patch('openedx.features.enterprise_support.api.Registry')
@mock.patch('openedx.features.enterprise_support.api.enterprise_customer_for_request')
def test_unlink_enterprise_user_from_idp_no_customer_user(self, mock_customer_from_request, mock_registry):
customer_idp = EnterpriseCustomerIdentityProviderFactory.create(
provider_id='the-provider',
)
customer = customer_idp.enterprise_customer
mock_customer_from_request.return_value = {
'uuid': customer.uuid,
}
mock_registry.get_enabled_by_backend_name.return_value = [
mock.Mock(provider_id='the-provider')
]
request = mock.Mock()
unlink_enterprise_user_from_idp(request, self.user, idp_backend_name='the-backend-name')
assert 0 == EnterpriseCustomerUser.objects.filter(user_id=self.user.id).count()