Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions cms/djangoapps/contentstore/views/preview.py
Original file line number Diff line number Diff line change
Expand Up @@ -207,15 +207,18 @@ def _preview_module_system(request, descriptor, field_data):
wrappers=wrappers,
wrappers_asides=wrappers_asides,
error_descriptor_class=ErrorBlock,
get_user_role=lambda: get_user_role(request.user, course_id),
# Get the raw DescriptorSystem, not the CombinedSystem
descriptor_runtime=descriptor._runtime, # pylint: disable=protected-access
services={
"field-data": field_data,
"i18n": ModuleI18nService,
'mako': mako_service,
"settings": SettingsService(),
"user": DjangoXBlockUserService(request.user, anonymous_user_id='student'),
"user": DjangoXBlockUserService(
request.user,
anonymous_user_id='student',
user_role=get_user_role(request.user, course_id),
),
"partitions": StudioPartitionService(course_id=course_id),
"teams_configuration": TeamsConfigurationService(),
},
Expand Down
10 changes: 10 additions & 0 deletions cms/djangoapps/contentstore/views/tests/test_preview.py
Original file line number Diff line number Diff line change
Expand Up @@ -230,6 +230,16 @@ def setUp(self):
self.request = RequestFactory().get('/dummy-url')
self.request.user = self.user
self.request.session = {}
self.descriptor = ItemFactory(category="video", parent=self.course)
self.field_data = mock.Mock()
self.runtime = _preview_module_system(
self.request,
self.descriptor,
self.field_data,
)

def test_get_user_role(self):
assert self.runtime.get_user_role() == 'staff'

@XBlock.register_temp_plugin(PureXBlock, identifier='pure')
def test_render_template(self):
Expand Down
2 changes: 2 additions & 0 deletions common/djangoapps/xblock_django/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,10 @@

# Optional attributes stored on the XBlockUser
ATTR_KEY_ANONYMOUS_USER_ID = 'edx-platform.anonymous_user_id'
ATTR_KEY_REQUEST_COUNTRY_CODE = 'edx-platform.request_country_code'
ATTR_KEY_IS_AUTHENTICATED = 'edx-platform.is_authenticated'
ATTR_KEY_USER_ID = 'edx-platform.user_id'
ATTR_KEY_USERNAME = 'edx-platform.username'
ATTR_KEY_USER_IS_STAFF = 'edx-platform.user_is_staff'
ATTR_KEY_USER_PREFERENCES = 'edx-platform.user_preferences'
ATTR_KEY_USER_ROLE = 'edx-platform.user_role'
76 changes: 66 additions & 10 deletions common/djangoapps/xblock_django/tests/test_user_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,11 @@
from common.djangoapps.xblock_django.user_service import (
ATTR_KEY_IS_AUTHENTICATED,
ATTR_KEY_ANONYMOUS_USER_ID,
ATTR_KEY_REQUEST_COUNTRY_CODE,
ATTR_KEY_USER_ID,
ATTR_KEY_USER_IS_STAFF,
ATTR_KEY_USER_PREFERENCES,
ATTR_KEY_USER_ROLE,
ATTR_KEY_USERNAME,
USER_PREFERENCES_WHITE_LIST,
DjangoXBlockUserService
Expand All @@ -37,15 +39,22 @@ def setUp(self):
set_user_preference(self.user, 'not_white_listed', 'hidden_value')
self.anon_user = AnonymousUserFactory()

def assert_is_anon_xb_user(self, xb_user):
def assert_is_anon_xb_user(self, xb_user, request_country_code):
"""
A set of assertions for an anonymous XBlockUser.
"""
assert not xb_user.opt_attrs[ATTR_KEY_IS_AUTHENTICATED]
assert xb_user.opt_attrs[ATTR_KEY_REQUEST_COUNTRY_CODE] == request_country_code
assert xb_user.full_name is None
self.assertListEqual(xb_user.emails, [])

def assert_xblock_user_matches_django(self, xb_user, dj_user, user_is_staff=False, anonymous_user_id=None):
def assert_xblock_user_matches_django(
self, xb_user, dj_user,
user_is_staff=False,
user_role=None,
anonymous_user_id=None,
request_country_code=None,
):
"""
A set of assertions for comparing a XBlockUser to a django User
"""
Expand All @@ -55,37 +64,49 @@ def assert_xblock_user_matches_django(self, xb_user, dj_user, user_is_staff=Fals
assert xb_user.opt_attrs[ATTR_KEY_USERNAME] == dj_user.username
assert xb_user.opt_attrs[ATTR_KEY_USER_ID] == dj_user.id
assert xb_user.opt_attrs[ATTR_KEY_USER_IS_STAFF] == user_is_staff
assert xb_user.opt_attrs[ATTR_KEY_USER_ROLE] == user_role
assert xb_user.opt_attrs[ATTR_KEY_ANONYMOUS_USER_ID] == anonymous_user_id
assert xb_user.opt_attrs[ATTR_KEY_REQUEST_COUNTRY_CODE] == request_country_code
assert all((pref in USER_PREFERENCES_WHITE_LIST) for pref in xb_user.opt_attrs[ATTR_KEY_USER_PREFERENCES])

def test_convert_anon_user(self):
"""
Tests for convert_django_user_to_xblock_user behavior when django user is AnonymousUser.
"""
django_user_service = DjangoXBlockUserService(self.anon_user)
country_code = 'UK'
django_user_service = DjangoXBlockUserService(self.anon_user, request_country_code=country_code)
xb_user = django_user_service.get_current_user()
assert xb_user.is_current_user
self.assert_is_anon_xb_user(xb_user)
self.assert_is_anon_xb_user(xb_user, request_country_code=country_code)

@ddt.data(
(False, None),
(True, None),
(False, 'abcdef0123'),
(True, 'abcdef0123'),
(False, None, None, None),
(True, 'instructor', None, None),
(True, 'staff', None, None),
(False, 'student', 'abcdef0123', None),
(True, 'student', 'abcdef0123', 'uk'),
)
@ddt.unpack
def test_convert_authenticate_user(self, user_is_staff, anonymous_user_id):
def test_convert_authenticate_user(self, user_is_staff, user_role, anonymous_user_id, request_country_code):
"""
Tests for convert_django_user_to_xblock_user behavior when django user is User.
"""
django_user_service = DjangoXBlockUserService(
self.user,
user_is_staff=user_is_staff,
user_role=user_role,
anonymous_user_id=anonymous_user_id,
request_country_code=request_country_code,
)
xb_user = django_user_service.get_current_user()
assert xb_user.is_current_user
self.assert_xblock_user_matches_django(xb_user, self.user, user_is_staff, anonymous_user_id)
self.assert_xblock_user_matches_django(
xb_user, self.user,
user_is_staff,
user_role,
anonymous_user_id,
request_country_code,
)

def test_get_anonymous_user_id_returns_none_for_non_staff_users(self):
"""
Expand Down Expand Up @@ -126,6 +147,27 @@ def test_get_anonymous_user_id_returns_id_for_existing_users(self):

assert anonymous_user_id == anon_user_id

def test_get_user_by_anonymous_id(self):
"""
Tests that get_user_by_anonymous_id returns the expected user.
"""
course_key = CourseKey.from_string('edX/toy/2012_Fall')
anon_user_id = anonymous_id_for_user(
user=self.user,
course_id=course_key
)

django_user_service = DjangoXBlockUserService(self.user)
user = django_user_service.get_user_by_anonymous_id(anon_user_id)
assert user == self.user

def test_get_user_by_anonymous_id_not_found(self):
"""
Tests that get_user_by_anonymous_id returns None for an unassigned anonymous user id.
"""
django_user_service = DjangoXBlockUserService(self.user)
assert django_user_service.get_user_by_anonymous_id('invalid-anon-id') is None

def test_external_id(self):
"""
Tests that external ids differ based on type.
Expand All @@ -138,3 +180,17 @@ def test_external_id(self):
assert ext_id1 != ext_id2
with pytest.raises(ValueError):
django_user_service.get_external_user_id('unknown')

def test_get_user_by_anonymous_id_assume_id(self):
"""
Tests that get_user_by_anonymous_id uses the anonymous user ID given to the service if none is provided.
"""
course_key = CourseKey.from_string('edX/toy/2012_Fall')
anon_user_id = anonymous_id_for_user(
user=self.user,
course_id=course_key
)

django_user_service = DjangoXBlockUserService(self.user, anonymous_user_id=anon_user_id)
user = django_user_service.get_user_by_anonymous_id()
assert user == self.user
21 changes: 20 additions & 1 deletion common/djangoapps/xblock_django/user_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,15 +9,17 @@

from openedx.core.djangoapps.external_user_ids.models import ExternalId
from openedx.core.djangoapps.user_api.preferences.api import get_user_preferences
from common.djangoapps.student.models import anonymous_id_for_user, get_user_by_username_or_email
from common.djangoapps.student.models import anonymous_id_for_user, get_user_by_username_or_email, user_by_anonymous_id

from .constants import (
ATTR_KEY_ANONYMOUS_USER_ID,
ATTR_KEY_IS_AUTHENTICATED,
ATTR_KEY_REQUEST_COUNTRY_CODE,
ATTR_KEY_USER_ID,
ATTR_KEY_USERNAME,
ATTR_KEY_USER_IS_STAFF,
ATTR_KEY_USER_PREFERENCES,
ATTR_KEY_USER_ROLE,
)


Expand All @@ -34,12 +36,16 @@ def __init__(self, django_user, **kwargs):

Args:
user_is_staff(bool): optional - whether the user is staff in the course
user_role(str): optional -- user's role in the course ('staff', 'instructor', or 'student')
anonymous_user_id(str): optional - anonymous_user_id for the user in the course
request_country_code(str): optional -- country code determined from the user's request IP address.
"""
super().__init__(**kwargs)
self._django_user = django_user
self._user_is_staff = kwargs.get('user_is_staff', False)
self._user_role = kwargs.get('user_role', 'student')
self._anonymous_user_id = kwargs.get('anonymous_user_id', None)
self._request_country_code = kwargs.get('request_country_code', None)

def get_current_user(self):
"""
Expand Down Expand Up @@ -80,6 +86,16 @@ def get_anonymous_user_id(self, username, course_id):
course_id = CourseKey.from_string(course_id)
return anonymous_id_for_user(user=user, course_id=course_id)

def get_user_by_anonymous_id(self, uid=None):
"""
Returns the Django User object corresponding to the given anonymous user id.

Returns None if there is no user with the given anonymous user id.

If no `uid` is provided, then the current anonymous user ID is used.
"""
Comment thread
Agrendalath marked this conversation as resolved.
Outdated
return user_by_anonymous_id(uid or self._anonymous_user_id)

def _convert_django_user_to_xblock_user(self, django_user):
"""
A function that returns an XBlockUser from the current Django request.user
Expand All @@ -96,9 +112,11 @@ def _convert_django_user_to_xblock_user(self, django_user):
xblock_user.emails = [django_user.email]
xblock_user.opt_attrs[ATTR_KEY_ANONYMOUS_USER_ID] = self._anonymous_user_id
xblock_user.opt_attrs[ATTR_KEY_IS_AUTHENTICATED] = True
xblock_user.opt_attrs[ATTR_KEY_REQUEST_COUNTRY_CODE] = self._request_country_code
xblock_user.opt_attrs[ATTR_KEY_USER_ID] = django_user.id
xblock_user.opt_attrs[ATTR_KEY_USERNAME] = django_user.username
xblock_user.opt_attrs[ATTR_KEY_USER_IS_STAFF] = self._user_is_staff
xblock_user.opt_attrs[ATTR_KEY_USER_ROLE] = self._user_role
user_preferences = get_user_preferences(django_user)
xblock_user.opt_attrs[ATTR_KEY_USER_PREFERENCES] = {
pref: user_preferences.get(pref)
Expand All @@ -107,5 +125,6 @@ def _convert_django_user_to_xblock_user(self, django_user):
}
else:
xblock_user.opt_attrs[ATTR_KEY_IS_AUTHENTICATED] = False
xblock_user.opt_attrs[ATTR_KEY_REQUEST_COUNTRY_CODE] = self._request_country_code

return xblock_user
2 changes: 1 addition & 1 deletion common/lib/xmodule/xmodule/lti_2_util.py
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ def lti_2_0_result_rest_handler(self, request, suffix):
except LTIError:
return Response(status=401) # Unauthorized in this case. 401 is right

real_user = self.system.get_real_user(anon_id)
real_user = self.system.service(self, 'user').get_user_by_anonymous_id(anon_id)
if not real_user: # that means we can't save to database, as we do not have real user id.
msg = f"[LTI]: Real user not found against anon_id: {anon_id}"
log.info(msg)
Expand Down
30 changes: 16 additions & 14 deletions common/lib/xmodule/xmodule/lti_module.py
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,10 @@
from openedx.core.djangolib.markup import HTML, Text
from xmodule.editing_module import EditingMixin

from common.djangoapps.xblock_django.constants import ATTR_KEY_ANONYMOUS_USER_ID
from common.djangoapps.xblock_django.constants import (
ATTR_KEY_ANONYMOUS_USER_ID,
ATTR_KEY_USER_ROLE,
)
from xmodule.lti_2_util import LTI20BlockMixin, LTIError
from xmodule.raw_module import EmptyDataRawMixin
from xmodule.util.xmodule_django import add_webpack_to_fragment
Expand Down Expand Up @@ -626,7 +629,8 @@ def role(self):
'staff': 'Administrator',
'instructor': 'Instructor',
}
return roles.get(self.system.get_user_role(), 'Student')
user_role = self.runtime.service(self, 'user').get_current_user().opt_attrs.get(ATTR_KEY_USER_ROLE)
return roles.get(user_role, 'Student')

def get_icon_class(self):
""" Returns the icon class """
Expand Down Expand Up @@ -676,17 +680,15 @@ def oauth_params(self, custom_parameters, client_key, client_secret):
# Username and email can't be sent in studio mode, because the user object is not defined.
# To test functionality test in LMS

if callable(self.runtime.get_real_user):
user_id = self.runtime.service(self, 'user').get_current_user().opt_attrs.get(ATTR_KEY_ANONYMOUS_USER_ID)
real_user_object = self.runtime.get_real_user(user_id)
try:
self.user_email = real_user_object.email # lint-amnesty, pylint: disable=attribute-defined-outside-init
except AttributeError:
self.user_email = "" # lint-amnesty, pylint: disable=attribute-defined-outside-init
try:
self.user_username = real_user_object.username # lint-amnesty, pylint: disable=attribute-defined-outside-init
except AttributeError:
self.user_username = "" # lint-amnesty, pylint: disable=attribute-defined-outside-init
real_user_object = self.runtime.service(self, 'user').get_user_by_anonymous_id()
try:
self.user_email = real_user_object.email # lint-amnesty, pylint: disable=attribute-defined-outside-init
except AttributeError:
self.user_email = "" # lint-amnesty, pylint: disable=attribute-defined-outside-init
try:
self.user_username = real_user_object.username # lint-amnesty, pylint: disable=attribute-defined-outside-init
except AttributeError:
self.user_username = "" # lint-amnesty, pylint: disable=attribute-defined-outside-init

if self.ask_to_send_username and self.user_username:
body["lis_person_sourcedid"] = self.user_username
Expand Down Expand Up @@ -837,7 +839,7 @@ def grade_handler(self, request, suffix): # lint-amnesty, pylint: disable=unuse
log.debug("[LTI]: " + error_message) # lint-amnesty, pylint: disable=logging-not-lazy
return Response(response_xml_template.format(**failure_values), content_type="application/xml")

real_user = self.system.get_real_user(parse.unquote(sourcedId.split(':')[-1]))
real_user = self.runtime.service(self, 'user').get_user_by_anonymous_id(parse.unquote(sourcedId.split(':')[-1]))
if not real_user: # that means we can't save to database, as we do not have real user id.
failure_values['imsx_messageIdentifier'] = escape(imsx_messageIdentifier)
failure_values['imsx_description'] = "User not found."
Expand Down
8 changes: 5 additions & 3 deletions common/lib/xmodule/xmodule/tests/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,7 @@ def get_test_system(
course_id=CourseKey.from_string('/'.join(['org', 'course', 'run'])),
user=None,
user_is_staff=False,
user_location=None,
render_template=None,
):
"""
Expand All @@ -102,10 +103,14 @@ def get_test_system(
"""
if not user:
user = Mock(name='get_test_system.user', is_staff=False)
if not user_location:
user_location = Mock(name='get_test_system.user_location')
user_service = StubUserService(
user=user,
anonymous_user_id='student',
user_is_staff=user_is_staff,
user_role='student',
request_country_code=user_location,
)

mako_service = StubMakoService(render_template=render_template)
Expand Down Expand Up @@ -133,7 +138,6 @@ def get_module(descriptor):
track_function=Mock(name='get_test_system.track_function'),
get_module=get_module,
replace_urls=str,
get_real_user=lambda __: user,
filestore=Mock(name='get_test_system.filestore', root_path='.'),
debug=True,
hostname="edx.org",
Expand All @@ -151,8 +155,6 @@ def get_module(descriptor):
node_path=os.environ.get("NODE_PATH", "/usr/local/lib/node_modules"),
course_id=course_id,
error_descriptor_class=ErrorBlock,
get_user_role=Mock(name='get_test_system.get_user_role', is_staff=False),
user_location=Mock(name='get_test_system.user_location'),
descriptor_runtime=descriptor_system,
)

Expand Down
Loading