diff --git a/eox_nelp/api_schema.py b/eox_nelp/api_schema.py deleted file mode 100644 index 8063f5e7..00000000 --- a/eox_nelp/api_schema.py +++ /dev/null @@ -1,65 +0,0 @@ -# """ -# Swagger view generator -# """ -# from django.conf import settings -# from django.conf.urls import url -# from drf_yasg.generators import OpenAPISchemaGenerator -# from drf_yasg.openapi import SwaggerDict -# from drf_yasg.views import get_schema_view -# from edx_api_doc_tools import get_docs_cache_timeout, make_api_info -# from rest_framework import permissions - -# from eox_nelp.api.v1 import views - - -# class APISchemaGenerator(OpenAPISchemaGenerator): -# """ -# Schema generator for eox-nelp. - -# Define an exclusive base path to perform requests for each endpoint and a -# specific security definition using oauth without overwritting project wide -# settings. -# """ - -# def get_schema(self, request=None, public=False): -# schema = super().get_schema(request, public) -# schema.base_path = '/eox-nelp/api/v1/' -# return schema - -# def get_security_definitions(self): -# security_definitions = { -# 'OAuth2': { -# 'flow': 'application', -# 'tokenUrl': f'{settings.LMS_ROOT_URL}/oauth2/access_token/', -# 'type': 'oauth2', -# }, -# } -# security_definitions = SwaggerDict.as_odict(security_definitions) -# return security_definitions - - -# api_urls = [ # pylint: disable=invalid-name -# url(r'^enrollment/$', views.EdxappEnrollment.as_view(), name='edxapp-enrollment'), -# url(r'^grade/$', views.EdxappGrade.as_view(), name='edxapp-grade'), -# url(r'^user/$', views.EdxappUser.as_view(), name='edxapp-user'), -# ] - -# if getattr(settings, "EOX_nelp_ENABLE_UPDATE_USERS", None): -# api_urls += [ -# url(r'^update-user/$', views.EdxappUserUpdater.as_view(), name='edxapp-user-updater'), -# ] - -# api_info = make_api_info( # pylint: disable=invalid-name -# title="eox nelp", -# version="v1", -# email=" contact@edunext.co", -# description="REST APIs to interact with edxapp", -# ) - -# docs_ui_view = get_schema_view( # pylint: disable=invalid-name -# api_info, -# generator_class=APISchemaGenerator, -# public=True, -# permission_classes=(permissions.AllowAny,), -# patterns=api_urls, -# ).with_ui('swagger', cache_timeout=get_docs_cache_timeout()) diff --git a/eox_nelp/apps.py b/eox_nelp/apps.py index e95bb1ea..e1e4d3a6 100644 --- a/eox_nelp/apps.py +++ b/eox_nelp/apps.py @@ -37,6 +37,7 @@ class eoxNelpConfig(AppConfig): 'common': {'relative_path': 'settings.common'}, 'test': {'relative_path': 'settings.test'}, 'production': {'relative_path': 'settings.production'}, + 'devstack': {'relative_path': 'settings.devstack'}, }, }, } diff --git a/eox_nelp/edxapp_wrapper/backends/__init__.py b/eox_nelp/edxapp_wrapper/backends/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/eox_nelp/edxapp_wrapper/backends/bearer_authentication_j_v1.py b/eox_nelp/edxapp_wrapper/backends/bearer_authentication_j_v1.py deleted file mode 100644 index 6e4b55f9..00000000 --- a/eox_nelp/edxapp_wrapper/backends/bearer_authentication_j_v1.py +++ /dev/null @@ -1,16 +0,0 @@ -"""Backend for authentication. - -This file contains all the necessary authentication dependencies from -https://github.com/eduNEXT/edunext-platform/tree/master/openedx/core/lib/api/authentication.py -""" -from openedx.core.lib.api.authentication import BearerAuthentication # pylint: disable=import-error - - -def get_bearer_authentication(): - """Allow to get the function BearerAuthentication from - https://github.com/eduNEXT/edunext-platform/tree/master/openedx/core/lib/api/authentication.py - - Returns: - BearerAuthentication function. - """ - return BearerAuthentication diff --git a/eox_nelp/edxapp_wrapper/backends/bearer_authentication_j_v1_test.py b/eox_nelp/edxapp_wrapper/backends/bearer_authentication_j_v1_test.py deleted file mode 100644 index 2576c090..00000000 --- a/eox_nelp/edxapp_wrapper/backends/bearer_authentication_j_v1_test.py +++ /dev/null @@ -1,15 +0,0 @@ -""" Backend test abstraction. """ - - -def get_bearer_authentication(): - """Allow to get the function BearerAuthentication from - https://github.com/eduNEXT/edunext-platform/tree/master/openedx/core/lib/api/authentication.py - - Returns: - BearerAuthentication function. - """ - try: - from openedx.core.lib.api.authentication import BearerAuthentication # pylint: disable=import-outside-toplevel - except ImportError: - BearerAuthentication = object - return BearerAuthentication diff --git a/eox_nelp/edxapp_wrapper/backends/certificates_h_v1.py b/eox_nelp/edxapp_wrapper/backends/certificates_h_v1.py deleted file mode 100644 index 10f3fb40..00000000 --- a/eox_nelp/edxapp_wrapper/backends/certificates_h_v1.py +++ /dev/null @@ -1,10 +0,0 @@ -""" -Backend for certificates app. -""" - -from lms.djangoapps.certificates.models import GeneratedCertificate # pylint: disable=import-error - - -def get_generated_certificate(): - """ get GeneratedCertificate model. """ - return GeneratedCertificate diff --git a/eox_nelp/edxapp_wrapper/backends/certificates_h_v1_test.py b/eox_nelp/edxapp_wrapper/backends/certificates_h_v1_test.py deleted file mode 100644 index 9a60f53e..00000000 --- a/eox_nelp/edxapp_wrapper/backends/certificates_h_v1_test.py +++ /dev/null @@ -1,15 +0,0 @@ -""" -Test backend to get GenerateCertificates Model. -""" - -from django.contrib.auth.models import Permission - - -def get_generated_certificate(): - """ - Get test GeneratedCertificate model. - - We return any django model that already exists so that - django-filters is happy and no migrations are created. - """ - return Permission diff --git a/eox_nelp/edxapp_wrapper/backends/comments_service_users_j_v1.py b/eox_nelp/edxapp_wrapper/backends/comments_service_users_j_v1.py deleted file mode 100644 index 347f688b..00000000 --- a/eox_nelp/edxapp_wrapper/backends/comments_service_users_j_v1.py +++ /dev/null @@ -1,17 +0,0 @@ -""" Module for the cs_comments_service User object.""" -from openedx.core.djangoapps.django_comment_common.comment_client.user import User # pylint: disable=import-error - - -def replace_username_cs_user(*args, **kwargs): - """ - Replace user's username in cs_comments_service (forums). - - kwargs: - user: edxapp user whose username is being replaced. - new_username: new username. - """ - user = kwargs.get("user") - new_username = kwargs.get("new_username") - - cs_user = User.from_django_user(user) - cs_user.replace_username(new_username) diff --git a/eox_nelp/edxapp_wrapper/backends/configuration_helpers_h_v1.py b/eox_nelp/edxapp_wrapper/backends/configuration_helpers_h_v1.py deleted file mode 100644 index 579bf132..00000000 --- a/eox_nelp/edxapp_wrapper/backends/configuration_helpers_h_v1.py +++ /dev/null @@ -1,7 +0,0 @@ -""" Backend abstraction. """ -from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers # pylint: disable=import-error - - -def get_configuration_helper(): - """ Backend to get the configuration helper. """ - return configuration_helpers diff --git a/eox_nelp/edxapp_wrapper/backends/configuration_helpers_h_v1_test.py b/eox_nelp/edxapp_wrapper/backends/configuration_helpers_h_v1_test.py deleted file mode 100644 index af636862..00000000 --- a/eox_nelp/edxapp_wrapper/backends/configuration_helpers_h_v1_test.py +++ /dev/null @@ -1,11 +0,0 @@ -""" Backend test abstraction. """ - - -def get_configuration_helper(): - """ Backend to get the configuration helper. """ - try: - from openedx.core.djangoapps.site_configuration import \ - helpers as configuration_helpers # pylint: disable=import-outside-toplevel - except ImportError: - configuration_helpers = object - return configuration_helpers diff --git a/eox_nelp/edxapp_wrapper/backends/coursekey_h_v1.py b/eox_nelp/edxapp_wrapper/backends/coursekey_h_v1.py deleted file mode 100644 index 5758d8fc..00000000 --- a/eox_nelp/edxapp_wrapper/backends/coursekey_h_v1.py +++ /dev/null @@ -1,51 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -Backend for the CourseKey validations that works under the open-release/hawthorn.beta1 tag -""" -# pylint: disable=import-error, protected-access -from __future__ import absolute_import, unicode_literals - -from django.conf import settings -from opaque_keys import InvalidKeyError -from opaque_keys.edx.keys import CourseKey -from rest_framework.serializers import ValidationError - -try: - from openedx.core.djangoapps.site_configuration.helpers import get_all_orgs, get_current_site_orgs -except ImportError: - get_all_orgs, get_current_site_orgs = object, object # pylint: disable=invalid-name - - -def get_valid_course_key(course_id): - """ - Return the CourseKey if the course_id is valid - """ - try: - return CourseKey.from_string(course_id) - except InvalidKeyError: - raise ValidationError(f"Invalid course_id {course_id}") from InvalidKeyError - - -def validate_org(course_id): - """ - Validate the course organization against all possible orgs for the site - - To determine if the Org is valid we must look at 3 things - 1 Orgs in the current site - 2 Orgs in other sites - 3 flag EOX_CORE_USER_ENABLE_MULTI_TENANCY - """ - - if not settings.EOX_CORE_USER_ENABLE_MULTI_TENANCY: - return True - - course_key = get_valid_course_key(course_id) - current_site_orgs = get_current_site_orgs() or [] - - if not current_site_orgs: # pylint: disable=no-else-return - if course_key.org in get_all_orgs(): - return False - return True - else: - return course_key.org in current_site_orgs diff --git a/eox_nelp/edxapp_wrapper/backends/courses_h_v1.py b/eox_nelp/edxapp_wrapper/backends/courses_h_v1.py deleted file mode 100644 index a8bbfa63..00000000 --- a/eox_nelp/edxapp_wrapper/backends/courses_h_v1.py +++ /dev/null @@ -1,56 +0,0 @@ -""" -Backend for contenstore courses. -""" -try: - from contentstore.views.course import ( # pylint: disable=import-error - _process_courses_list, - get_courses_accessible_to_user, - ) - from openedx.core.djangoapps.content.course_overviews.models import CourseOverview # pylint: disable=import-error - from openedx.core.djangoapps.models.course_details import CourseDetails # pylint: disable=import-error -except ImportError: - # pylint: disable=invalid-name - _process_courses_list = object - get_courses_accessible_to_user = object - CourseOverview = object - CourseDetails = object - - -def courses_accessible_to_user(*args, **kwargs): - """ get get_courses_accessible_to_user function. """ - return get_courses_accessible_to_user(*args, **kwargs) - - -def get_process_courses_list(*args, **kwargs): - """ get _process_courses_list function. """ - return _process_courses_list(*args, **kwargs) - - -def get_course_details_fields(): - """ get CourseDetails fields. """ - course_details = CourseDetails(org=None, course_id=None, run=None) - all_fields = vars(course_details) - - return all_fields.keys() - - -def get_first_course_key(): - """ - Returns the course key of any course in order to get - the advance settings keys to populate the course settings - tab in the Course management view. - """ - - course = CourseOverview.objects.all()[:1] - if course: - course_location = course[0].location - else: - return '' - return course_location.course_key - - -def get_course_overview(): - """ - Gets course overview model from edxapp. - """ - return CourseOverview diff --git a/eox_nelp/edxapp_wrapper/backends/courseware_h_v1.py b/eox_nelp/edxapp_wrapper/backends/courseware_h_v1.py deleted file mode 100644 index 225af19c..00000000 --- a/eox_nelp/edxapp_wrapper/backends/courseware_h_v1.py +++ /dev/null @@ -1,10 +0,0 @@ -""" -Backend for courseware module. -""" - -from lms.djangoapps.courseware import courses # pylint: disable=import-error - - -def get_courseware_courses(): - """ get courses. """ - return courses diff --git a/eox_nelp/edxapp_wrapper/backends/edxfuture_i_v1.py b/eox_nelp/edxapp_wrapper/backends/edxfuture_i_v1.py deleted file mode 100644 index 864f44b9..00000000 --- a/eox_nelp/edxapp_wrapper/backends/edxfuture_i_v1.py +++ /dev/null @@ -1,39 +0,0 @@ -""" -Functions copied from a version higer than hawthorn (for backwards compatibility with it) -Must be deleted some day and replaced with calls to the actual functions -""" -from django.conf import settings -# pylint: disable=import-error -from django.core.cache import cache -from openedx.core.djangoapps.catalog.models import CatalogIntegration -from openedx.core.djangoapps.catalog.utils import create_catalog_api_client - - -def get_program(program_uuid, ignore_cache=False): - """ - Retrieves the details for the specified program. - - Args: - program_uuid (UUID): Program identifier - ignore_cache (bool): Indicates if previously-cached data should be ignored. - - Returns: - dict - """ - program_uuid = str(program_uuid) - cache_key = f'programs.api.data.{program_uuid}' - - if not ignore_cache: - program = cache.get(cache_key) - - if program: - return program - - catalog_integration = CatalogIntegration.current() - user = catalog_integration.get_service_user() - api = create_catalog_api_client(user) - - program = api.programs(program_uuid).get() - cache.set(cache_key, program, getattr(settings, 'PROGRAMS_CACHE_TTL', 60)) - - return program diff --git a/eox_nelp/edxapp_wrapper/backends/enrollment_h_v1.py b/eox_nelp/edxapp_wrapper/backends/enrollment_h_v1.py deleted file mode 100644 index 8caa87d6..00000000 --- a/eox_nelp/edxapp_wrapper/backends/enrollment_h_v1.py +++ /dev/null @@ -1,351 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -Backend for the create_edxapp_user that works under the open-release/hawthorn.beta1 tag -""" -# pylint: disable=import-error, protected-access -from __future__ import absolute_import, unicode_literals - -import datetime -import logging - -from course_modes.models import CourseMode -from django.contrib.auth.models import User -from opaque_keys import InvalidKeyError -from opaque_keys.edx.keys import CourseKey -from openedx.core.djangoapps.content.course_overviews.models import CourseOverview -from openedx.core.lib.exceptions import CourseNotFoundError -from pytz import utc -from rest_framework.exceptions import APIException, NotFound -from student.models import CourseEnrollment - -from eox_core.edxapp_wrapper.backends.edxfuture_i_v1 import get_program -from eox_core.edxapp_wrapper.coursekey import get_valid_course_key, validate_org -from eox_core.edxapp_wrapper.users import check_edxapp_account_conflicts - -try: - # For Hawthorn and Ironwood versions. - from enrollment import api - from enrollment.errors import CourseEnrollmentExistsError, CourseModeNotFoundError -except ImportError: - # For Juniper versions. - from openedx.core.djangoapps.enrollments import api # pylint: disable=ungrouped-imports - from openedx.core.djangoapps.enrollments.errors import ( # pylint: disable=ungrouped-imports - CourseEnrollmentExistsError, - CourseModeNotFoundError, - ) - - -LOG = logging.getLogger(__name__) - - -def create_enrollment(user, *args, **kwargs): - """ - backend function to create enrollment - - Example: - >>>create_enrollment( - user_object, - { - "course_id": "course-v1-edX-DemoX-1T2015", - ... - } - ) - - """ - kwargs = dict(kwargs) - program_uuid = kwargs.pop('bundle_id', None) - course_id = kwargs.pop('course_id', None) - - if program_uuid: - return _enroll_on_program(user, program_uuid, *args, **kwargs) - if course_id: - return _enroll_on_course(user, course_id, *args, **kwargs) - - raise APIException("You have to provide a course_id or bundle_id") - - -def update_enrollment(user, course_id, mode, *args, **kwargs): - """ - Update enrollment of given user in the course provided. - - Example: - >>>update_enrollment( - user_object, - course_id, - mode, - is_active=False, - enrollment_attributes=[ - { - "namespace": "credit", - "name": "provider_id", - "value": "hogwarts", - }, - {...} - ] - } - ) - """ - username = user.username - - is_active = kwargs.get('is_active', True) - enrollment_attributes = kwargs.get('enrollment_attributes', None) - - LOG.info('Updating enrollment for student: %s of course: %s mode: %s', username, course_id, mode) - enrollment = api._data_api().update_course_enrollment(username, course_id, mode, is_active) - if not enrollment: - raise NotFound(f'No enrollment found for {username}') - if enrollment_attributes is not None: - api.set_enrollment_attributes(username, course_id, enrollment_attributes) - - return { - 'user': enrollment['user'], - 'course_id': course_id, - 'mode': enrollment['mode'], - 'is_active': enrollment['is_active'], - 'enrollment_attributes': enrollment_attributes, - } - - -def get_enrollment(*args, **kwargs): - """ - Return enrollment of given user in the course provided. - - Example: - >>>get_enrollment( - { - "username": "Bob", - "course_id": "course-v1-edX-DemoX-1T2015" - } - ) - """ - errors = [] - course_id = kwargs.pop('course_id', None) - username = kwargs.get('username', None) - - try: - LOG.info('Getting enrollment information of student: %s course: %s', username, course_id) - enrollment = api.get_enrollment(username, course_id) - if not enrollment: - errors.append(f'No enrollment found for user:`{username}`') - return None, errors - except InvalidKeyError: - errors.append(f'No course found for course_id `{course_id}`') - return None, errors - enrollment['enrollment_attributes'] = api.get_enrollment_attributes(username, course_id) - enrollment['course_id'] = course_id - return enrollment, errors - - -def delete_enrollment(*args, **kwargs): - """ - Delete enrollment and enrollment attributes of given user in the course provided. - - Example: - >>>delete_enrollment( - { - "user": user_object, - "course_id": course-v1-edX-DemoX-1T2015" - ) - """ - course_id = kwargs.pop('course_id', None) - user = kwargs.get('user') - try: - course_key = get_valid_course_key(course_id) - except InvalidKeyError: - raise NotFound(f'No course found by course id `{course_id}`') from InvalidKeyError - - username = user.username - - LOG.info('Deleting enrollment. User: `%s` course: `%s`', username, course_id) - enrollment = CourseEnrollment.get_enrollment(user, course_key) - if not enrollment: - raise NotFound(f'No enrollment found for user: `{username}` on course_id `{course_id}`') - try: - enrollment.delete() - except Exception: - raise NotFound(f'Error: Enrollment could not be deleted for user: `{username}` on course_id `{course_id}`') from Exception - - -def _enroll_on_course(user, course_id, *args, **kwargs): - """ - enroll user on a single course - - Example: - >>>_enroll_on_course( - { - "user": user_object, - "course_id": course-v1-edX-DemoX-1T2015", - "is_active": "False", - "mode": "audit", - "enrollment_attributes": [ - { - "namespace": "credit", - "name": "provider_id", - "value": "hogwarts", - }, - {...} - ] - } - ) - """ - errors = [] - - username = user.username - - mode = kwargs.get('mode', 'audit') - is_active = kwargs.get('is_active', True) - force = kwargs.get('force', False) - enrollment_attributes = kwargs.get('enrollment_attributes', None) - - enrollment_valid_query = { - 'course_id': course_id, - 'force': force, - 'mode': mode, - 'username': username, - } - validation_errors = check_edxapp_enrollment_is_valid(**enrollment_valid_query) - if validation_errors: - return None, [", ".join(validation_errors)] - - try: - LOG.info('Creating regular enrollment %s, %s, %s', username, course_id, mode) - enrollment = _create_or_update_enrollment(username, course_id, mode, is_active, force) - except CourseNotFoundError as err: - raise NotFound(repr(err)) from err - except Exception as err: # pylint: disable=broad-except - if force: - LOG.info('Force create enrollment %s, %s, %s', username, course_id, mode) - enrollment = _force_create_enrollment(username, course_id, mode, is_active) - else: - if not str(err): - err_msg = err.__class__.__name__ - raise APIException(detail=err_msg if err_msg else err) from err - - if enrollment_attributes is not None: - api.set_enrollment_attributes(username, course_id, enrollment_attributes) - try: - enrollment['enrollment_attributes'] = enrollment_attributes - enrollment['course_id'] = course_id - except TypeError: - pass - return enrollment, errors - - -def _enroll_on_program(user, program_uuid, *arg, **kwargs): - """ - enroll user on each of the courses of a program - """ - results = [] - errors = [] - LOG.info('Enrolling on program: %s', program_uuid) - try: - data = get_program(program_uuid) - except Exception as err: # pylint: disable=broad-except - raise NotFound(repr(err)) from err - if not data['courses']: - raise NotFound("No courses found for this program") - for course in data['courses']: - if course['course_runs']: - course_run = _get_preferred_course_run(course) - LOG.info('Enrolling on course_run: %s', course_run['key']) - course_id = course_run['key'] - try: - result, errors_list = _enroll_on_course(user, course_id, *arg, **kwargs) - except APIException as error: - result = { - 'username': user.username, - 'mode': None, - 'course_id': course_id, - } - errors_list = [error.detail] - - results.append(result) - errors.append(errors_list) - else: - raise NotFound("No course runs available for this course") - return results, errors - - -def _get_preferred_course_run(course): - """ - Returns the course run more likely to be the intended one - """ - sorted_course_runs = sorted(course['course_runs'], key=lambda run: run['start']) - - for run in sorted_course_runs: - default_enrollment_start_date = datetime.datetime(1900, 1, 1, tzinfo=utc) - course_run_key = CourseKey.from_string(run['key']) - course_overview = CourseOverview.get_from_id(course_run_key) - enrollment_end = course_overview.enrollment_end or datetime.datetime.max.replace(tzinfo=utc) - enrollment_start = course_overview.enrollment_start or default_enrollment_start_date - run['is_enrollment_open'] = enrollment_start <= datetime.datetime.now(utc) < enrollment_end - - open_course_runs = [run for run in sorted_course_runs if run['is_enrollment_open']] - course_run = open_course_runs[0] if open_course_runs else sorted_course_runs[-1] - return course_run - - -# pylint: disable=invalid-name -def check_edxapp_enrollment_is_valid(*args, **kwargs): - """ - backend function to check if enrollment is valid - """ - errors = [] - is_active = kwargs.get("is_active", True) - course_id = kwargs.get("course_id") - force = kwargs.get('force', False) - mode = kwargs.get("mode") - program_uuid = kwargs.get('bundle_id') - username = kwargs.get("username") - email = kwargs.get("email") - - if program_uuid and course_id: - return ['You have to provide a course_id or bundle_id but not both'] - if not program_uuid and not course_id: - return ['You have to provide a course_id or bundle_id'] - if not email and not username: - return ['Email or username needed'] - if not check_edxapp_account_conflicts(email=email, username=username): - return ['User not found'] - if mode not in CourseMode.ALL_MODES: - return ['Invalid mode given:' + mode] - if course_id: - if not validate_org(course_id): - errors.append('Enrollment not allowed for given org') - if course_id and not force: - try: - api.validate_course_mode(course_id, mode, is_active=is_active) - except CourseModeNotFoundError: - errors.append('Mode not found') - except CourseNotFoundError: - errors.append('Course not found') - return errors - - -def _create_or_update_enrollment(username, course_id, mode, is_active, try_update): - """ - non-forced create or update enrollment internal function - """ - try: - enrollment = api._data_api().create_course_enrollment(username, course_id, mode, is_active) - except CourseEnrollmentExistsError as err: - if try_update: - enrollment = api._data_api().update_course_enrollment(username, course_id, mode, is_active) - else: - raise Exception(repr(err) + ", use force to update the existing enrollment") from err - return enrollment - - -def _force_create_enrollment(username, course_id, mode, is_active): - """ - forced create enrollment internal function - """ - try: - course_key = get_valid_course_key(course_id) - user = User.objects.get(username=username) - enrollment = CourseEnrollment.enroll(user, course_key, check_access=False) - api._data_api()._update_enrollment(enrollment, is_active=is_active, mode=mode) - except Exception as err: # pylint: disable=broad-except - raise APIException(repr(err)) from err - return enrollment diff --git a/eox_nelp/edxapp_wrapper/backends/enrollment_l_v1.py b/eox_nelp/edxapp_wrapper/backends/enrollment_l_v1.py deleted file mode 100644 index 1f8aaada..00000000 --- a/eox_nelp/edxapp_wrapper/backends/enrollment_l_v1.py +++ /dev/null @@ -1,341 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -Backend for the create_edxapp_user that works under the open-release/lilac.master tag -""" -# pylint: disable=import-error, protected-access -import datetime -import logging - -from common.djangoapps.course_modes.models import CourseMode -from common.djangoapps.student.models import CourseEnrollment -from django.contrib.auth.models import User -from opaque_keys import InvalidKeyError -from opaque_keys.edx.keys import CourseKey -from openedx.core.djangoapps.content.course_overviews.models import CourseOverview -from openedx.core.djangoapps.enrollments import api # pylint: disable=ungrouped-imports -from openedx.core.djangoapps.enrollments.errors import ( # pylint: disable=ungrouped-imports - CourseEnrollmentExistsError, - CourseModeNotFoundError, -) -from openedx.core.lib.exceptions import CourseNotFoundError -from pytz import utc -from rest_framework.exceptions import APIException, NotFound - -from eox_core.edxapp_wrapper.backends.edxfuture_i_v1 import get_program -from eox_core.edxapp_wrapper.coursekey import get_valid_course_key, validate_org -from eox_core.edxapp_wrapper.users import check_edxapp_account_conflicts - -LOG = logging.getLogger(__name__) - - -def create_enrollment(user, *args, **kwargs): - """ - backend function to create enrollment - - Example: - >>>create_enrollment( - user_object, - { - "course_id": "course-v1-edX-DemoX-1T2015", - ... - } - ) - - """ - kwargs = dict(kwargs) - program_uuid = kwargs.pop('bundle_id', None) - course_id = kwargs.pop('course_id', None) - - if program_uuid: - return _enroll_on_program(user, program_uuid, *args, **kwargs) - if course_id: - return _enroll_on_course(user, course_id, *args, **kwargs) - - raise APIException("You have to provide a course_id or bundle_id") - - -def update_enrollment(user, course_id, mode, *args, **kwargs): - """ - Update enrollment of given user in the course provided. - - Example: - >>>update_enrollment( - user_object, - course_id, - mode, - is_active=False, - enrollment_attributes=[ - { - "namespace": "credit", - "name": "provider_id", - "value": "hogwarts", - }, - {...} - ] - } - ) - """ - username = user.username - - is_active = kwargs.get('is_active', True) - enrollment_attributes = kwargs.get('enrollment_attributes', None) - - LOG.info('Updating enrollment for student: %s of course: %s mode: %s', username, course_id, mode) - enrollment = api._data_api().update_course_enrollment(username, course_id, mode, is_active) - if not enrollment: - raise NotFound(f'No enrollment found for {username}') - if enrollment_attributes is not None: - api.set_enrollment_attributes(username, course_id, enrollment_attributes) - - return { - 'user': enrollment['user'], - 'course_id': course_id, - 'mode': enrollment['mode'], - 'is_active': enrollment['is_active'], - 'enrollment_attributes': enrollment_attributes, - } - - -def get_enrollment(*args, **kwargs): - """ - Return enrollment of given user in the course provided. - - Example: - >>>get_enrollment( - { - "username": "Bob", - "course_id": "course-v1-edX-DemoX-1T2015" - } - ) - """ - errors = [] - course_id = kwargs.pop('course_id', None) - username = kwargs.get('username', None) - - try: - LOG.info('Getting enrollment information of student: %s course: %s', username, course_id) - enrollment = api.get_enrollment(username, course_id) - if not enrollment: - errors.append(f'No enrollment found for user:`{username}`') - return None, errors - except InvalidKeyError: - errors.append(f'No course found for course_id `{course_id}`') - return None, errors - enrollment['enrollment_attributes'] = api.get_enrollment_attributes(username, course_id) - enrollment['course_id'] = course_id - return enrollment, errors - - -def delete_enrollment(*args, **kwargs): - """ - Delete enrollment and enrollment attributes of given user in the course provided. - - Example: - >>>delete_enrollment( - { - "user": user_object, - "course_id": course-v1-edX-DemoX-1T2015" - ) - """ - course_id = kwargs.pop('course_id', None) - user = kwargs.get('user') - try: - course_key = get_valid_course_key(course_id) - except InvalidKeyError: - raise NotFound(f'No course found by course id `{course_id}`') from InvalidKeyError - - username = user.username - - LOG.info('Deleting enrollment. User: `%s` course: `%s`', username, course_id) - enrollment = CourseEnrollment.get_enrollment(user, course_key) - if not enrollment: - raise NotFound(f'No enrollment found for user: `{username}` on course_id `{course_id}`') - try: - enrollment.delete() - except Exception: - raise NotFound(f'Error: Enrollment could not be deleted for user: `{username}` on course_id `{course_id}`') from Exception - - -def _enroll_on_course(user, course_id, *args, **kwargs): - """ - enroll user on a single course - - Example: - >>>_enroll_on_course( - { - "user": user_object, - "course_id": course-v1-edX-DemoX-1T2015", - "is_active": "False", - "mode": "audit", - "enrollment_attributes": [ - { - "namespace": "credit", - "name": "provider_id", - "value": "hogwarts", - }, - {...} - ] - } - ) - """ - errors = [] - - username = user.username - - mode = kwargs.get('mode', 'audit') - is_active = kwargs.get('is_active', True) - force = kwargs.get('force', False) - enrollment_attributes = kwargs.get('enrollment_attributes', None) - - enrollment_valid_query = { - 'course_id': course_id, - 'force': force, - 'mode': mode, - 'username': username, - } - validation_errors = check_edxapp_enrollment_is_valid(**enrollment_valid_query) - if validation_errors: - return None, [", ".join(validation_errors)] - - try: - LOG.info('Creating regular enrollment %s, %s, %s', username, course_id, mode) - enrollment = _create_or_update_enrollment(username, course_id, mode, is_active, force) - except CourseNotFoundError as err: - raise NotFound(repr(err)) from err - except Exception as err: # pylint: disable=broad-except - if force: - LOG.info('Force create enrollment %s, %s, %s', username, course_id, mode) - enrollment = _force_create_enrollment(username, course_id, mode, is_active) - else: - if not str(err): - err_msg = err.__class__.__name__ - raise APIException(detail=err_msg if err_msg else err) from err - - if enrollment_attributes is not None: - api.set_enrollment_attributes(username, course_id, enrollment_attributes) - try: - enrollment['enrollment_attributes'] = enrollment_attributes - enrollment['course_id'] = course_id - except TypeError: - pass - return enrollment, errors - - -def _enroll_on_program(user, program_uuid, *arg, **kwargs): - """ - enroll user on each of the courses of a program - """ - results = [] - errors = [] - LOG.info('Enrolling on program: %s', program_uuid) - try: - data = get_program(program_uuid) - except Exception as err: # pylint: disable=broad-except - raise NotFound(repr(err)) from err - if not data['courses']: - raise NotFound("No courses found for this program") - for course in data['courses']: - if course['course_runs']: - course_run = _get_preferred_course_run(course) - LOG.info('Enrolling on course_run: %s', course_run['key']) - course_id = course_run['key'] - try: - result, errors_list = _enroll_on_course(user, course_id, *arg, **kwargs) - except APIException as error: - result = { - 'username': user.username, - 'mode': None, - 'course_id': course_id, - } - errors_list = [error.detail] - - results.append(result) - errors.append(errors_list) - else: - raise NotFound("No course runs available for this course") - return results, errors - - -def _get_preferred_course_run(course): - """ - Returns the course run more likely to be the intended one - """ - sorted_course_runs = sorted(course['course_runs'], key=lambda run: run['start']) - - for run in sorted_course_runs: - default_enrollment_start_date = datetime.datetime(1900, 1, 1, tzinfo=utc) - course_run_key = CourseKey.from_string(run['key']) - course_overview = CourseOverview.get_from_id(course_run_key) - enrollment_end = course_overview.enrollment_end or datetime.datetime.max.replace(tzinfo=utc) - enrollment_start = course_overview.enrollment_start or default_enrollment_start_date - run['is_enrollment_open'] = enrollment_start <= datetime.datetime.now(utc) < enrollment_end - - open_course_runs = [run for run in sorted_course_runs if run['is_enrollment_open']] - course_run = open_course_runs[0] if open_course_runs else sorted_course_runs[-1] - return course_run - - -# pylint: disable=invalid-name -def check_edxapp_enrollment_is_valid(*args, **kwargs): - """ - backend function to check if enrollment is valid - """ - errors = [] - is_active = kwargs.get("is_active", True) - course_id = kwargs.get("course_id") - force = kwargs.get('force', False) - mode = kwargs.get("mode") - program_uuid = kwargs.get('bundle_id') - username = kwargs.get("username") - email = kwargs.get("email") - - if program_uuid and course_id: - return ['You have to provide a course_id or bundle_id but not both'] - if not program_uuid and not course_id: - return ['You have to provide a course_id or bundle_id'] - if not email and not username: - return ['Email or username needed'] - if not check_edxapp_account_conflicts(email=email, username=username): - return ['User not found'] - if mode not in CourseMode.ALL_MODES: - return ['Invalid mode given:' + mode] - if course_id: - if not validate_org(course_id): - errors.append('Enrollment not allowed for given org') - if course_id and not force: - try: - api.validate_course_mode(course_id, mode, is_active=is_active) - except CourseModeNotFoundError: - errors.append('Mode not found') - except CourseNotFoundError: - errors.append('Course not found') - return errors - - -def _create_or_update_enrollment(username, course_id, mode, is_active, try_update): - """ - non-forced create or update enrollment internal function - """ - try: - enrollment = api._data_api().create_course_enrollment(username, course_id, mode, is_active) - except CourseEnrollmentExistsError as err: - if try_update: - enrollment = api._data_api().update_course_enrollment(username, course_id, mode, is_active) - else: - raise Exception(repr(err) + ", use force to update the existing enrollment") from err - return enrollment - - -def _force_create_enrollment(username, course_id, mode, is_active): - """ - forced create enrollment internal function - """ - try: - course_key = get_valid_course_key(course_id) - user = User.objects.get(username=username) - enrollment = CourseEnrollment.enroll(user, course_key, check_access=False) - api._data_api()._update_enrollment(enrollment, is_active=is_active, mode=mode) - except Exception as err: # pylint: disable=broad-except - raise APIException(repr(err)) from err - return enrollment diff --git a/eox_nelp/edxapp_wrapper/backends/grades_h_v1.py b/eox_nelp/edxapp_wrapper/backends/grades_h_v1.py deleted file mode 100644 index f9235347..00000000 --- a/eox_nelp/edxapp_wrapper/backends/grades_h_v1.py +++ /dev/null @@ -1,10 +0,0 @@ -""" -Backend for grades. -""" - -from lms.djangoapps.grades.course_grade_factory import CourseGradeFactory # pylint: disable=import-error - - -def get_course_grade_factory(): - """ get CourseGradeFactory object. """ - return CourseGradeFactory diff --git a/eox_nelp/edxapp_wrapper/backends/pre_enrollment_h_v1.py b/eox_nelp/edxapp_wrapper/backends/pre_enrollment_h_v1.py deleted file mode 100644 index a23cb812..00000000 --- a/eox_nelp/edxapp_wrapper/backends/pre_enrollment_h_v1.py +++ /dev/null @@ -1,117 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -Backend for the pre-enrollment (white listings) functionality that works under the open-release/hawthorn.beta1 tag -""" -# pylint: disable=import-error, protected-access -from __future__ import absolute_import, unicode_literals - -import logging - -from django.db import IntegrityError -from rest_framework.exceptions import NotFound -from student.models import CourseEnrollmentAllowed - -from eox_core.edxapp_wrapper.coursekey import get_valid_course_key -from eox_core.edxapp_wrapper.courseware import get_courseware_courses - -LOG = logging.getLogger(__name__) - - -def create_pre_enrollment(*args, **kwargs): - """ - Create pre-enrollment of given user in the course provided. - - Example: - >>>create_pre_enrollment( - { - "email": "bob@example.com", - "course_id": course-v1-edX-DemoX-1T2015", - "auto_enroll": "False" - } - ) - """ - warnings = [] - email = kwargs.get('email') - auto_enroll = kwargs.get('auto_enroll', False) - course_id = kwargs.pop('course_id') - - try: - course_key = get_valid_course_key(course_id) - pre_enrollment = CourseEnrollmentAllowed.objects.create(course_id=course_key, **kwargs) - # Check if the course exists otherwise add a warning - course = get_courseware_courses().get_course(course_key) - LOG.info('Creating regular pre-enrollment for email: %s course_id: %s auto_enroll: %s', email, course.id, auto_enroll) - except IntegrityError: - pre_enrollment = None - raise NotFound(f'Pre-enrollment already exists for email: {email} course_id: {course_id}') from IntegrityError - except ValueError: - warnings = [f'Course with course_id:{course_id} does not exist'] - return pre_enrollment, warnings - - -def update_pre_enrollment(*args, **kwargs): - """ - Update pre-enrollment of given user in the course provided. - - Example: - >>>update_pre_enrollment( - { - "email": "bob@example.com", - "course_id": course-v1-edX-DemoX-1T2015", - "auto_enroll": "False" - } - ) - """ - auto_enroll = kwargs.pop('auto_enroll', False) - pre_enrollment = kwargs.get('pre_enrollment') - try: - pre_enrollment.auto_enroll = auto_enroll - pre_enrollment.save() - LOG.info('Updating regular pre-enrollment for email: %s course_id: %s auto_enroll: %s', pre_enrollment.email, pre_enrollment.course_id, auto_enroll) - except Exception: # pylint: disable=broad-except - raise NotFound(f'Pre-enrollment not found for email: {pre_enrollment.email} course_id: {pre_enrollment.course_id}') from Exception - return pre_enrollment - - -def delete_pre_enrollment(*args, **kwargs): - """ - Delete pre-enrollment of given user in the course provided. - - Example: - >>>delete_pre_enrollment( - { - "email": "bob@example.com", - "course_id": course-v1-edX-DemoX-1T2015" - } - ) - """ - pre_enrollment = kwargs.get('pre_enrollment') - try: - LOG.info('Deleting regular pre-enrollment for email: %s course_id: %s', pre_enrollment.email, pre_enrollment.course_id) - pre_enrollment.delete() - except Exception: # pylint: disable=broad-except - raise NotFound(f'Pre-enrollment not found for email: {pre_enrollment.email} course_id: {pre_enrollment.course_id}') from Exception - - -def get_pre_enrollment(*args, **kwargs): - """ - Get pre-enrollment of given user in the course provided. - - Example: - >>>get_pre_enrollment( - { - "email": "bob@example.com", - "course_id": course-v1-edX-DemoX-1T2015" - } - ) - """ - email = kwargs.get('email') - course_id = kwargs.pop('course_id') - try: - course_key = get_valid_course_key(course_id) - pre_enrollment = CourseEnrollmentAllowed.objects.get(course_id=course_key, email=email) - LOG.info('Getting regular pre-enrollment for email: %s course_id: %s', email, course_id) - except CourseEnrollmentAllowed.DoesNotExist: - raise NotFound(f'Pre-enrollment not found for email: {email} course_id: {course_id}') from CourseEnrollmentAllowed.DoesNotExist - return pre_enrollment diff --git a/eox_nelp/edxapp_wrapper/backends/pre_enrollment_l_v1.py b/eox_nelp/edxapp_wrapper/backends/pre_enrollment_l_v1.py deleted file mode 100644 index 8ea2c1b0..00000000 --- a/eox_nelp/edxapp_wrapper/backends/pre_enrollment_l_v1.py +++ /dev/null @@ -1,115 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -Backend for the pre-enrollment (white listings) functionality that works under the open-release/lilac.master tag -""" -# pylint: disable=import-error, protected-access -import logging - -from common.djangoapps.student.models import CourseEnrollmentAllowed # pylint: disable=no-name-in-module -from django.db import IntegrityError -from rest_framework.exceptions import NotFound - -from eox_core.edxapp_wrapper.coursekey import get_valid_course_key -from eox_core.edxapp_wrapper.courseware import get_courseware_courses - -LOG = logging.getLogger(__name__) - - -def create_pre_enrollment(*args, **kwargs): - """ - Create pre-enrollment of given user in the course provided. - - Example: - >>>create_pre_enrollment( - { - "email": "bob@example.com", - "course_id": course-v1-edX-DemoX-1T2015", - "auto_enroll": "False" - } - ) - """ - warnings = [] - email = kwargs.get('email') - auto_enroll = kwargs.get('auto_enroll', False) - course_id = kwargs.pop('course_id') - - try: - course_key = get_valid_course_key(course_id) - pre_enrollment = CourseEnrollmentAllowed.objects.create(course_id=course_key, **kwargs) - # Check if the course exists otherwise add a warning - course = get_courseware_courses().get_course(course_key) - LOG.info('Creating regular pre-enrollment for email: %s course_id: %s auto_enroll: %s', email, course.id, auto_enroll) - except IntegrityError: - pre_enrollment = None - raise NotFound(f'Pre-enrollment already exists for email: {email} course_id: {course_id}') from IntegrityError - except ValueError: - warnings = [f'Course with course_id:{course_id} does not exist'] - return pre_enrollment, warnings - - -def update_pre_enrollment(*args, **kwargs): - """ - Update pre-enrollment of given user in the course provided. - - Example: - >>>update_pre_enrollment( - { - "email": "bob@example.com", - "course_id": course-v1-edX-DemoX-1T2015", - "auto_enroll": "False" - } - ) - """ - auto_enroll = kwargs.pop('auto_enroll', False) - pre_enrollment = kwargs.get('pre_enrollment') - try: - pre_enrollment.auto_enroll = auto_enroll - pre_enrollment.save() - LOG.info('Updating regular pre-enrollment for email: %s course_id: %s auto_enroll: %s', pre_enrollment.email, pre_enrollment.course_id, auto_enroll) - except Exception: # pylint: disable=broad-except - raise NotFound(f'Pre-enrollment not found for email: {pre_enrollment.email} course_id: {pre_enrollment.course_id}') from Exception - return pre_enrollment - - -def delete_pre_enrollment(*args, **kwargs): - """ - Delete pre-enrollment of given user in the course provided. - - Example: - >>>delete_pre_enrollment( - { - "email": "bob@example.com", - "course_id": course-v1-edX-DemoX-1T2015" - } - ) - """ - pre_enrollment = kwargs.get('pre_enrollment') - try: - LOG.info('Deleting regular pre-enrollment for email: %s course_id: %s', pre_enrollment.email, pre_enrollment.course_id) - pre_enrollment.delete() - except Exception: # pylint: disable=broad-except - raise NotFound(f'Pre-enrollment not found for email: {pre_enrollment.email} course_id: {pre_enrollment.course_id}') from Exception - - -def get_pre_enrollment(*args, **kwargs): - """ - Get pre-enrollment of given user in the course provided. - - Example: - >>>get_pre_enrollment( - { - "email": "bob@example.com", - "course_id": course-v1-edX-DemoX-1T2015" - } - ) - """ - email = kwargs.get('email') - course_id = kwargs.pop('course_id') - try: - course_key = get_valid_course_key(course_id) - pre_enrollment = CourseEnrollmentAllowed.objects.get(course_id=course_key, email=email) - LOG.info('Getting regular pre-enrollment for email: %s course_id: %s', email, course_id) - except CourseEnrollmentAllowed.DoesNotExist: - raise NotFound(f'Pre-enrollment not found for email: {email} course_id: {course_id}') from CourseEnrollmentAllowed.DoesNotExist - return pre_enrollment diff --git a/eox_nelp/edxapp_wrapper/backends/storages_i_v1.py b/eox_nelp/edxapp_wrapper/backends/storages_i_v1.py deleted file mode 100644 index 9f17022e..00000000 --- a/eox_nelp/edxapp_wrapper/backends/storages_i_v1.py +++ /dev/null @@ -1,20 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -Storages backend -""" -from openedx.core.storage import DevelopmentStorage, ProductionStorage # pylint: disable=import-error - - -def get_edxapp_production_staticfiles_storage(): # pylint: disable=invalid-name - """ - Return the edx-platform production staticfiles storage - """ - return ProductionStorage - - -def get_edxapp_development_staticfiles_storage(): # pylint: disable=invalid-name - """ - Return the edx-platform development staticfiles storage - """ - return DevelopmentStorage diff --git a/eox_nelp/edxapp_wrapper/backends/storages_i_v1_test.py b/eox_nelp/edxapp_wrapper/backends/storages_i_v1_test.py deleted file mode 100644 index 1bb3b938..00000000 --- a/eox_nelp/edxapp_wrapper/backends/storages_i_v1_test.py +++ /dev/null @@ -1,28 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -Storages test backend -""" -from eox_core.test_utils import TestStorage - - -def get_edxapp_production_staticfiles_storage(): # pylint: disable=invalid-name - """ - Return the edx-platform production staticfiles storage - """ - try: - from openedx.core.storage import ProductionStorage # pylint: disable=import-outside-toplevel - except ImportError: - ProductionStorage = TestStorage - return ProductionStorage - - -def get_edxapp_development_staticfiles_storage(): # pylint: disable=invalid-name - """ - Return the edx-platform development staticfiles storage - """ - try: - from openedx.core.storage import DevelopmentStorage # pylint: disable=import-outside-toplevel - except ImportError: - DevelopmentStorage = TestStorage - return DevelopmentStorage diff --git a/eox_nelp/edxapp_wrapper/backends/third_party_auth_j_v1.py b/eox_nelp/edxapp_wrapper/backends/third_party_auth_j_v1.py deleted file mode 100644 index 85b6d5e1..00000000 --- a/eox_nelp/edxapp_wrapper/backends/third_party_auth_j_v1.py +++ /dev/null @@ -1,11 +0,0 @@ -"""Backend for the third party authentication exception middleware.""" - - -def get_tpa_exception_middleware(): - """Get the ExceptionMiddleware class.""" - try: - from third_party_auth.middleware import ExceptionMiddleware # pylint: disable=import-outside-toplevel - except ImportError: - from django.utils.deprecation import \ - MiddlewareMixin as ExceptionMiddleware # pylint: disable=import-outside-toplevel - return ExceptionMiddleware diff --git a/eox_nelp/edxapp_wrapper/backends/third_party_auth_l_v1.py b/eox_nelp/edxapp_wrapper/backends/third_party_auth_l_v1.py deleted file mode 100644 index 23cebd70..00000000 --- a/eox_nelp/edxapp_wrapper/backends/third_party_auth_l_v1.py +++ /dev/null @@ -1,12 +0,0 @@ -"""Backend for the third party authentication exception middleware.""" - - -def get_tpa_exception_middleware(): - """Get the ExceptionMiddleware class.""" - try: - from common.djangoapps.third_party_auth.middleware import \ - ExceptionMiddleware # pylint: disable=import-outside-toplevel - except ImportError: - from django.utils.deprecation import \ - MiddlewareMixin as ExceptionMiddleware # pylint: disable=import-outside-toplevel - return ExceptionMiddleware diff --git a/eox_nelp/edxapp_wrapper/backends/users_h_v1.py b/eox_nelp/edxapp_wrapper/backends/users_h_v1.py deleted file mode 100644 index 14c1f2c5..00000000 --- a/eox_nelp/edxapp_wrapper/backends/users_h_v1.py +++ /dev/null @@ -1,225 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -Backend for the create_edxapp_user that works under the open-release/hawthorn.beta1 tag -""" -from __future__ import absolute_import, unicode_literals - -import logging - -from django.conf import settings -from django.contrib.auth import get_user_model -from django.db import transaction -from openedx.core.djangoapps.lang_pref import LANGUAGE_KEY # pylint: disable=import-error -from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers # pylint: disable=import-error -from openedx.core.djangoapps.user_api.accounts import USERNAME_MAX_LENGTH # pylint: disable=import-error,unused-import -from openedx.core.djangoapps.user_api.accounts.api import check_account_exists # pylint: disable=import-error -from openedx.core.djangoapps.user_api.accounts.serializers import UserReadOnlySerializer # pylint: disable=import-error -from openedx.core.djangoapps.user_api.preferences import api as preferences_api # pylint: disable=import-error -from rest_framework.exceptions import NotFound -from student.forms import AccountCreationForm # pylint: disable=import-error -from student.helpers import create_or_set_user_attribute_created_on_site # pylint: disable=import-error -from student.helpers import do_create_account # pylint: disable=import-error; pylint: disable=import-error -from student.models import CourseEnrollment # pylint: disable=import-error; pylint: disable=import-error -from student.models import ( # pylint: disable=import-error - UserAttribute, - UserProfile, - UserSignupSource, - create_comments_service_user, -) - -LOG = logging.getLogger(__name__) -User = get_user_model() # pylint: disable=invalid-name - - -def get_user_read_only_serializer(): - """ - Great serializer that fits our needs - """ - return UserReadOnlySerializer - - -def check_edxapp_account_conflicts(email, username): - """ - Exposed function to check conflicts - """ - return check_account_exists(email=email, username=username) - - -def create_edxapp_user(*args, **kwargs): - """ - Creates a user on the open edx django site using calls to - functions defined in the edx-platform codebase - - Example call: - - data = { - 'email': "address@example.org", - 'username': "Username", - 'password': "P4ssW0rd", - 'fullname': "Full Name", - 'activate': True, - 'site': request.site, - 'language_preference': 'es-419', - } - user = create_edxapp_user(**data) - - """ - errors = [] - - email = kwargs.pop("email") - username = kwargs.pop("username") - conflicts = check_edxapp_account_conflicts(email=email, username=username) - if conflicts: - return None, [f"Fatal: account collition with the provided: {', '.join(conflicts)}"] - - data = { - 'username': username, - 'email': email, - 'password': kwargs.pop("password"), - 'name': kwargs.pop("fullname"), - } - # Go ahead and create the new user - with transaction.atomic(): - # In theory is possible to extend the registration form with a custom app - # An example form app for this can be found at http://github.com/open-craft/custom-form-app - # form = get_registration_extension_form(data=params) - # if not form: - form = AccountCreationForm( - data=data, - tos_required=False, - # TODO: we need to support the extra profile fields as defined in the django.settings - # extra_fields=extra_fields, - # extended_profile_fields=extended_profile_fields, - # enforce_password_policy=enforce_password_policy, - ) - (user, profile, registration) = do_create_account(form) # pylint: disable=unused-variable - - site = kwargs.pop("site", False) - if site: - create_or_set_user_attribute_created_on_site(user, site) - else: - errors.append("The user was not assigned to any site") - - try: - create_comments_service_user(user) - except Exception: # pylint: disable=broad-except - errors.append("No comments_service_user was created") - - # TODO: link account with third party auth - - lang_pref = kwargs.pop("language_preference", False) - if lang_pref: - try: - preferences_api.set_user_preference(user, LANGUAGE_KEY, lang_pref) - except Exception: # pylint: disable=broad-except - errors.append(f"Could not set lang preference '{lang_pref}' for user '{user.username}'") - - if kwargs.pop("activate_user", False): - user.is_active = True - user.save() - - # TODO: run conditional email sequence - - return user, errors - - -def get_edxapp_user(**kwargs): - """ - Retrieve an user by username and/or email - - The user will be returned only if it belongs to the calling site - - Examples: - >>> get_edxapp_user( - { - "username": "Bob", - "site": request.site - } - ) - >>> get_edxapp_user( - { - "email": "Bob@mailserver.com", - "site": request.site - } - ) - """ - params = {key: kwargs.get(key) for key in ['username', 'email'] if key in kwargs} - site = kwargs.get('site') - try: - domain = site.domain - except AttributeError: - domain = None - - try: - user = User.objects.get(**params) - for source_method in FetchUserSiteSources.get_enabled_source_methods(): - if source_method(user, domain): - break - else: - raise User.DoesNotExist - except User.DoesNotExist: - raise NotFound(f'No user found by {str(params)} on site {domain}.') from User.DoesNotExist - return user - - -def get_course_team_user(*args, **kwargs): - """ - Get _course_team_user function. - We need to check if the SERVICE_VARIANT is equal to cms, since - contentstore is a module registered in the INSTALLED_APPS - of the cms only. - """ - if settings.SERVICE_VARIANT == 'cms': - from contentstore.views.user import _course_team_user # pylint: disable=import-error,import-outside-toplevel - return _course_team_user(*args, **kwargs) - return None - - -class FetchUserSiteSources: - """ - Methods to make the comparison to check if an user belongs to a site plus the - get_enabled_source_methods that just brings an array of functions enabled to do so - """ - - @classmethod - def get_enabled_source_methods(cls): - """ Brings the array of methods to check if an user belongs to a site. """ - sources = configuration_helpers.get_value( - 'EOX_CORE_USER_ORIGIN_SITE_SOURCES', - getattr(settings, 'EOX_CORE_USER_ORIGIN_SITE_SOURCES') - ) - return [getattr(cls, source) for source in sources] - - @staticmethod - def fetch_from_created_on_site_prop(user, domain): - """ Fetch option. """ - if not domain: - return False - return UserAttribute.get_user_attribute(user, 'created_on_site') == domain - - @staticmethod - def fetch_from_user_signup_source(user, domain): - """ Read the signup source. """ - return len(UserSignupSource.objects.filter(user=user, site=domain)) > 0 - - @staticmethod - def fetch_from_unfiltered_table(user, site): - """ Fetch option that does not take into account the multi-tentancy model of the installation. """ - return bool(user) - - -def get_course_enrollment(): - """ get CourseEnrollment model """ - return CourseEnrollment - - -def get_user_signup_source(): - """ get UserSignupSource model """ - return UserSignupSource - - -def get_user_profile(): - """ Gets the UserProfile model """ - - return UserProfile diff --git a/eox_nelp/edxapp_wrapper/backends/users_h_v1_test.py b/eox_nelp/edxapp_wrapper/backends/users_h_v1_test.py deleted file mode 100644 index 98c2698d..00000000 --- a/eox_nelp/edxapp_wrapper/backends/users_h_v1_test.py +++ /dev/null @@ -1,102 +0,0 @@ -""" -Test backend to get CourseEnrollment Model. -""" - -from django.contrib.auth.models import Permission - -USERNAME_MAX_LENGTH = 30 - - -def get_edxapp_user(**kwargs): - """ - Return a fake user - """ - return object - - -def create_edxapp_user(*args, **kwargs): - """ - Return a fake user and a list of errors - """ - return object, [] - - -def get_user_read_only_serializer(): - """ - Return a fake user read only serializer - """ - try: - from openedx.core.djangoapps.user_api.accounts.serializers import \ - UserReadOnlySerializer # pylint: disable=import-outside-toplevel - except ImportError: - UserReadOnlySerializer = object - return UserReadOnlySerializer - - -def check_edxapp_account_conflicts(email, username): - """ - Get an executes the check_account_exists for tests - """ - try: - from openedx.core.djangoapps.user_api.accounts.api import \ - check_account_exists # pylint: disable=import-outside-toplevel - except ImportError: - def check_account_exists(email=None, username=None): - """ - Fake definition for check_account_exists edxapp function - """ - return email and username - return check_account_exists(email=email, username=username) - - -def get_course_enrollment(): - """ - Get Test CourseEnrollment model. - - We return any django model that already exists so that - django-filters is happy and no migrations are created. - """ - return Permission - - -def get_course_team_user(*args, **kwargs): - """ - Return a fake course team user - """ - return object - - -def get_user_signup_source(): - """ - Get test UserSignupSource model - """ - try: - from student.models import UserSignupSource # pylint: disable=import-outside-toplevel - except ImportError: - UserSignupSource = object - return UserSignupSource - - -def get_user_profile(): - """ Gets the UserProfile model """ - try: - from student.models import UserProfile # pylint: disable=import-outside-toplevel - except ImportError: - UserProfile = object - return UserProfile - - -def generate_password(*args, **kwargs): - """ Generates a password """ - return "ThisShouldBeARandomPassword" - - -def get_user_attribute(): - """ - Get test UserAttribute model - """ - try: - from student.models import UserAttribute # pylint: disable=import-outside-toplevel - except ImportError: - UserAttribute = object - return UserAttribute diff --git a/eox_nelp/edxapp_wrapper/backends/users_j_v1.py b/eox_nelp/edxapp_wrapper/backends/users_j_v1.py deleted file mode 100644 index 4b8a01a0..00000000 --- a/eox_nelp/edxapp_wrapper/backends/users_j_v1.py +++ /dev/null @@ -1,346 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -Backend for the create_edxapp_user that works under the open-release/juniper.master tag -""" -# pylint: disable=import-error -from __future__ import absolute_import, unicode_literals - -import logging - -from django import forms -from django.conf import settings -from django.contrib.auth import get_user_model -from django.db import transaction -from openedx.core.djangoapps.lang_pref import LANGUAGE_KEY -from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers -from openedx.core.djangoapps.user_api.accounts import USERNAME_MAX_LENGTH # pylint: disable=unused-import -from openedx.core.djangoapps.user_api.accounts.serializers import UserReadOnlySerializer -from openedx.core.djangoapps.user_api.accounts.views import _set_unusable_password # pylint: disable=unused-import -from openedx.core.djangoapps.user_api.models import UserRetirementStatus -from openedx.core.djangoapps.user_api.preferences import api as preferences_api -from openedx.core.djangoapps.user_authn.utils import generate_password # pylint: disable=unused-import -from openedx.core.djangoapps.user_authn.views.registration_form import AccountCreationForm -from openedx.core.djangolib.oauth2_retirement_utils import retire_dot_oauth2_models # pylint: disable=unused-import -from rest_framework import status -from rest_framework.exceptions import NotFound -from social_django.models import UserSocialAuth -from student.helpers import create_or_set_user_attribute_created_on_site, do_create_account -from student.models import ( - CourseEnrollment, - LoginFailures, - Registration, - UserAttribute, - UserProfile, - UserSignupSource, - create_comments_service_user, - email_exists_or_retired, - get_retired_email_by_email, - username_exists_or_retired, -) - -# pylint: disable=ungrouped-imports -try: - from openedx.core.lib.triggers.v1 import post_register -except ImportError: - # In case edx-platform -vanilla- Juniper release is used - from openedx.core.djangoapps.user_authn.views.register import REGISTER_USER as post_register - - -LOG = logging.getLogger(__name__) -User = get_user_model() # pylint: disable=invalid-name - - -def get_user_read_only_serializer(): - """ - Great serializer that fits our needs - """ - return UserReadOnlySerializer - - -def check_edxapp_account_conflicts(email, username): - """ - Exposed function to check conflicts - """ - conflicts = [] - - if username and username_exists_or_retired(username): - conflicts.append("username") - - if email and email_exists_or_retired(email): - conflicts.append("email") - - return conflicts - - -class EdnxAccountCreationForm(AccountCreationForm): - """ - A form to extend the behaviour of the AccountCreationForm. - For now the purpose of this form is to allow to make the - password optional if the flag 'skip_password' is True. - - This form it's currently only used for validation, not rendering. - """ - - def __init__( # pylint:disable=too-many-arguments - self, - data=None, - extra_fields=None, - extended_profile_fields=None, - do_third_party_auth=True, - tos_required=True, - ): - super().__init__( - data=data, - extra_fields=extra_fields, - extended_profile_fields=extended_profile_fields, - do_third_party_auth=do_third_party_auth, - tos_required=tos_required, - ) - - if data.pop("skip_password", False): - self.fields['password'] = forms.CharField(required=False) - - -def create_edxapp_user(*args, **kwargs): - """ - Creates a user on the open edx django site using calls to - functions defined in the edx-platform codebase - - Example call: - - data = { - 'email': "address@example.org", - 'username': "Username", - 'password': "P4ssW0rd", - 'fullname': "Full Name", - 'activate': True, - 'site': request.site, - 'language_preference': 'es-419', - } - user = create_edxapp_user(**data) - - """ - errors = [] - - extra_fields = getattr(settings, "REGISTRATION_EXTRA_FIELDS", {}) - extended_profile_fields = getattr(settings, "extended_profile_fields", []) - kwargs["name"] = kwargs.pop("fullname", None) - email = kwargs.get("email") - username = kwargs.get("username") - conflicts = check_edxapp_account_conflicts(email=email, username=username) - if conflicts: - return None, [f"Fatal: account collition with the provided: {', '.join(conflicts)}"] - - # Go ahead and create the new user - with transaction.atomic(): - # In theory is possible to extend the registration form with a custom app - # An example form app for this can be found at http://github.com/open-craft/custom-form-app - # form = get_registration_extension_form(data=params) - # if not form: - form = EdnxAccountCreationForm( - data=kwargs, - tos_required=False, - extra_fields=extra_fields, - extended_profile_fields=extended_profile_fields, - # enforce_password_policy=enforce_password_policy, - ) - (user, profile, registration) = do_create_account(form) # pylint: disable=unused-variable - - site = kwargs.pop("site", False) - if site: - create_or_set_user_attribute_created_on_site(user, site) - else: - errors.append("The user was not assigned to any site") - - try: - create_comments_service_user(user) - except Exception: # pylint: disable=broad-except - errors.append("No comments_service_user was created") - - # TODO: link account with third party auth - - # Announce registration through API call - post_register.send_robust(sender=None, user=user) # pylint: disable=no-member - - lang_pref = kwargs.pop("language_preference", False) - if lang_pref: - try: - preferences_api.set_user_preference(user, LANGUAGE_KEY, lang_pref) - except Exception: # pylint: disable=broad-except - errors.append(f"Could not set lang preference '{lang_pref}' for user '{user.username}'") - - if kwargs.pop("activate_user", False): - user.is_active = True - user.save() - - # TODO: run conditional email sequence - - return user, errors - - -def get_edxapp_user(**kwargs): - """ - Retrieve a user by username and/or email - - The user will be returned only if it belongs to the calling site - - Examples: - >>> get_edxapp_user( - { - "username": "Bob", - "site": request.site - } - ) - >>> get_edxapp_user( - { - "email": "Bob@mailserver.com", - "site": request.site - } - ) - """ - params = {key: kwargs.get(key) for key in ['username', 'email'] if key in kwargs} - site = kwargs.get('site') - try: - domain = site.domain - except AttributeError: - domain = None - - try: - user = User.objects.get(**params) - for source_method in FetchUserSiteSources.get_enabled_source_methods(): - if source_method(user, domain): - break - else: - raise User.DoesNotExist - except User.DoesNotExist: - raise NotFound(f'No user found by {str(params)} on site {domain}.') from User.DoesNotExist - return user - - -def delete_edxapp_user(*args, **kwargs): - """ - Deletes a user from the platform. - """ - msg = None - - user = kwargs.get("user") - case_id = kwargs.get("case_id") - site = kwargs.get("site") - is_support_user = kwargs.get("is_support_user") - - user_response = f"The user {user.username} <{user.email}> " - - signup_sources = user.usersignupsource_set.all() - sources = [signup_source.site for signup_source in signup_sources] - - if site and site.name.upper() in (source.upper() for source in sources): - if len(sources) == 1: - with transaction.atomic(): - support_label = "_support" if is_support_user else "" - user.email = f"{user.email}{case_id}.ednx{support_label}_retired" - user.save() - - # Add user to retirement queue. - UserRetirementStatus.create_retirement(user) - - # Unlink LMS social auth accounts - UserSocialAuth.objects.filter(user_id=user.id).delete() - - # Change LMS password & email - user.email = get_retired_email_by_email(user.email) - user.save() - _set_unusable_password(user) - - # Remove the activation keys sent by email to the user for account activation. - Registration.objects.filter(user=user).delete() - - # Delete OAuth tokens associated with the user. - retire_dot_oauth2_models(user) - - # Delete user signup source object - signup_sources[0].delete() - - msg = f"{user_response} has been removed" - else: - for signup_source in signup_sources: - if signup_source.site.upper() == site.name.upper(): - signup_source.delete() - - msg = f"{user_response} has more than one signup source. The signup source from the site {site} has been deleted" - - return msg, status.HTTP_200_OK - - raise NotFound(f"{user_response} does not have a signup source on the site {site}") - - -def get_course_team_user(*args, **kwargs): - """ - Get _course_team_user function. - We need to check if the SERVICE_VARIANT is equal to cms, since - contentstore is a module registered in the INSTALLED_APPS - of the cms only. - """ - if settings.SERVICE_VARIANT == 'cms': - from contentstore.views.user import _course_team_user # pylint: disable=import-error,import-outside-toplevel - return _course_team_user(*args, **kwargs) - return None - - -class FetchUserSiteSources: - """ - Methods to make the comparison to check if an user belongs to a site plus the - get_enabled_source_methods that just brings an array of functions enabled to do so - """ - - @classmethod - def get_enabled_source_methods(cls): - """ Brings the array of methods to check if an user belongs to a site. """ - sources = configuration_helpers.get_value( - 'EOX_CORE_USER_ORIGIN_SITE_SOURCES', - getattr(settings, 'EOX_CORE_USER_ORIGIN_SITE_SOURCES') - ) - return [getattr(cls, source) for source in sources] - - @staticmethod - def fetch_from_created_on_site_prop(user, domain): - """ Fetch option. """ - if not domain: - return False - return UserAttribute.get_user_attribute(user, 'created_on_site') == domain - - @staticmethod - def fetch_from_user_signup_source(user, domain): - """ Read the signup source. """ - return len(UserSignupSource.objects.filter(user=user, site=domain)) > 0 - - @staticmethod - def fetch_from_unfiltered_table(user, site): - """ Fetch option that does not take into account the multi-tentancy model of the installation. """ - return bool(user) - - -def get_course_enrollment(): - """ get CourseEnrollment model """ - return CourseEnrollment - - -def get_user_signup_source(): - """ get UserSignupSource model """ - return UserSignupSource - - -def get_login_failures(): - """ get LoginFailures model """ - return LoginFailures - - -def get_user_profile(): - """ Gets the UserProfile model """ - - return UserProfile - - -def get_user_attribute(): - """ Gets the UserAttribute model """ - return UserAttribute diff --git a/eox_nelp/edxapp_wrapper/backends/users_l_v1.py b/eox_nelp/edxapp_wrapper/backends/users_l_v1.py deleted file mode 100644 index 308685b4..00000000 --- a/eox_nelp/edxapp_wrapper/backends/users_l_v1.py +++ /dev/null @@ -1,357 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -Backend for the create_edxapp_user that works under the open-release/lilac.master tag -""" -import logging - -from common.djangoapps.student.helpers import ( # pylint: disable=import-error,no-name-in-module - create_or_set_user_attribute_created_on_site, - do_create_account, -) -from common.djangoapps.student.models import ( # pylint: disable=import-error,no-name-in-module - CourseEnrollment, - LoginFailures, - Registration, - UserAttribute, - UserProfile, - UserSignupSource, - create_comments_service_user, - email_exists_or_retired, - get_retired_email_by_email, - username_exists_or_retired, -) -from django import forms -from django.conf import settings -from django.contrib.auth import get_user_model -from django.db import transaction -from openedx.core.djangoapps.lang_pref import LANGUAGE_KEY # pylint: disable=import-error -from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers # pylint: disable=import-error -from openedx.core.djangoapps.user_api.accounts import USERNAME_MAX_LENGTH # pylint: disable=import-error,unused-import -from openedx.core.djangoapps.user_api.accounts.serializers import UserReadOnlySerializer # pylint: disable=import-error -from openedx.core.djangoapps.user_api.accounts.views import \ - _set_unusable_password # pylint: disable=import-error,unused-import -from openedx.core.djangoapps.user_api.models import UserRetirementStatus # pylint: disable=import-error -from openedx.core.djangoapps.user_api.preferences import api as preferences_api # pylint: disable=import-error -from openedx.core.djangoapps.user_authn.utils import generate_password # pylint: disable=import-error,unused-import -from openedx.core.djangoapps.user_authn.views.register import \ - REGISTER_USER as post_register # pylint: disable=import-error -from openedx.core.djangoapps.user_authn.views.registration_form import ( # pylint: disable=import-error - AccountCreationForm, -) -from openedx.core.djangolib.oauth2_retirement_utils import \ - retire_dot_oauth2_models # pylint: disable=import-error,unused-import -from openedx_events.learning.data import UserData, UserPersonalData # pylint: disable=import-error -from openedx_events.learning.signals import STUDENT_REGISTRATION_COMPLETED # pylint: disable=import-error -from rest_framework import status -from rest_framework.exceptions import NotFound -from social_django.models import UserSocialAuth # pylint: disable=import-error - -LOG = logging.getLogger(__name__) -User = get_user_model() # pylint: disable=invalid-name - - -def get_user_read_only_serializer(): - """ - Great serializer that fits our needs - """ - return UserReadOnlySerializer - - -def check_edxapp_account_conflicts(email, username): - """ - Exposed function to check conflicts - """ - conflicts = [] - - if username and username_exists_or_retired(username): - conflicts.append("username") - - if email and email_exists_or_retired(email): - conflicts.append("email") - - return conflicts - - -class EdnxAccountCreationForm(AccountCreationForm): - """ - A form to extend the behaviour of the AccountCreationForm. - For now the purpose of this form is to allow to make the - password optional if the flag 'skip_password' is True. - This form it's currently only used for validation, not rendering. - """ - - def __init__( # pylint:disable=too-many-arguments - self, - data=None, - extra_fields=None, - extended_profile_fields=None, - do_third_party_auth=True, - tos_required=True, - ): - super().__init__( - data=data, - extra_fields=extra_fields, - extended_profile_fields=extended_profile_fields, - do_third_party_auth=do_third_party_auth, - tos_required=tos_required, - ) - - if data.pop("skip_password", False): - self.fields['password'] = forms.CharField(required=False) - - -def create_edxapp_user(*args, **kwargs): - """ - Creates a user on the open edx django site using calls to - functions defined in the edx-platform codebase - - Example call: - - data = { - 'email': "address@example.org", - 'username': "Username", - 'password': "P4ssW0rd", - 'fullname': "Full Name", - 'activate': True, - 'site': request.site, - 'language_preference': 'es-419', - } - user = create_edxapp_user(**data) - - """ - errors = [] - - extra_fields = getattr(settings, "REGISTRATION_EXTRA_FIELDS", {}) - extended_profile_fields = getattr(settings, "extended_profile_fields", []) - kwargs["name"] = kwargs.pop("fullname", None) - email = kwargs.get("email") - username = kwargs.get("username") - conflicts = check_edxapp_account_conflicts(email=email, username=username) - if conflicts: - return None, [f"Fatal: account collition with the provided: {', '.join(conflicts)}"] - - # Go ahead and create the new user - with transaction.atomic(): - # In theory is possible to extend the registration form with a custom app - # An example form app for this can be found at http://github.com/open-craft/custom-form-app - # form = get_registration_extension_form(data=params) - # if not form: - form = EdnxAccountCreationForm( - data=kwargs, - tos_required=False, - extra_fields=extra_fields, - extended_profile_fields=extended_profile_fields, - # enforce_password_policy=enforce_password_policy, - ) - (user, profile, registration) = do_create_account(form) # pylint: disable=unused-variable - - site = kwargs.pop("site", False) - if site: - create_or_set_user_attribute_created_on_site(user, site) - else: - errors.append("The user was not assigned to any site") - - try: - create_comments_service_user(user) - except Exception: # pylint: disable=broad-except - errors.append("No comments_service_user was created") - - # TODO: link account with third party auth - - # Announce registration through API call - post_register.send_robust(sender=None, user=user) # pylint: disable=no-member - - STUDENT_REGISTRATION_COMPLETED.send_event( - user=UserData( - pii=UserPersonalData( - username=user.username, - email=user.email, - name=user.profile.name, # pylint: disable=no-member - ), - id=user.id, - is_active=user.is_active, - ), - ) - - lang_pref = kwargs.pop("language_preference", False) - if lang_pref: - try: - preferences_api.set_user_preference(user, LANGUAGE_KEY, lang_pref) - except Exception: # pylint: disable=broad-except - errors.append(f"Could not set lang preference '{lang_pref}' for user '{user.username}'") - - if kwargs.pop("activate_user", False): - user.is_active = True - user.save() - - # TODO: run conditional email sequence - - return user, errors - - -def get_edxapp_user(**kwargs): - """ - Retrieve a user by username and/or email - - The user will be returned only if it belongs to the calling site - - Examples: - >>> get_edxapp_user( - { - "username": "Bob", - "site": request.site - } - ) - >>> get_edxapp_user( - { - "email": "Bob@mailserver.com", - "site": request.site - } - ) - """ - params = {key: kwargs.get(key) for key in ['username', 'email'] if key in kwargs} - site = kwargs.get('site') - try: - domain = site.domain - except AttributeError: - domain = None - - try: - user = User.objects.get(**params) - for source_method in FetchUserSiteSources.get_enabled_source_methods(): - if source_method(user, domain): - break - else: - raise User.DoesNotExist - except User.DoesNotExist: - raise NotFound(f'No user found by {str(params)} on site {domain}.') from User.DoesNotExist - return user - - -def delete_edxapp_user(*args, **kwargs): - """ - Deletes a user from the platform. - """ - msg = None - - user = kwargs.get("user") - case_id = kwargs.get("case_id") - site = kwargs.get("site") - is_support_user = kwargs.get("is_support_user") - - user_response = f"The user {user.username} <{user.email}> " - - signup_sources = user.usersignupsource_set.all() - sources = [signup_source.site for signup_source in signup_sources] - - if site and site.name.upper() in (source.upper() for source in sources): - if len(sources) == 1: - with transaction.atomic(): - support_label = "_support" if is_support_user else "" - user.email = f"{user.email}{case_id}.ednx{support_label}_retired" - user.save() - - # Add user to retirement queue. - UserRetirementStatus.create_retirement(user) - - # Unlink LMS social auth accounts - UserSocialAuth.objects.filter(user_id=user.id).delete() - - # Change LMS password & email - user.email = get_retired_email_by_email(user.email) - user.save() - _set_unusable_password(user) - - # Remove the activation keys sent by email to the user for account activation. - Registration.objects.filter(user=user).delete() - - # Delete OAuth tokens associated with the user. - retire_dot_oauth2_models(user) - - # Delete user signup source object - signup_sources[0].delete() - - msg = f"{user_response} has been removed" - else: - for signup_source in signup_sources: - if signup_source.site.upper() == site.name.upper(): - signup_source.delete() - - msg = f"{user_response} has more than one signup source. The signup source from the site {site} has been deleted" - - return msg, status.HTTP_200_OK - - raise NotFound(f"{user_response} does not have a signup source on the site {site}") - - -def get_course_team_user(*args, **kwargs): - """ - Get _course_team_user function. - We need to check if the SERVICE_VARIANT is equal to cms, since - contentstore is a module registered in the INSTALLED_APPS - of the cms only. - """ - if settings.SERVICE_VARIANT == 'cms': - from contentstore.views.user import _course_team_user # pylint: disable=import-error,import-outside-toplevel - return _course_team_user(*args, **kwargs) - return None - - -class FetchUserSiteSources: - """ - Methods to make the comparison to check if an user belongs to a site plus the - get_enabled_source_methods that just brings an array of functions enabled to do so - """ - - @classmethod - def get_enabled_source_methods(cls): - """ Brings the array of methods to check if an user belongs to a site. """ - sources = configuration_helpers.get_value( - 'EOX_CORE_USER_ORIGIN_SITE_SOURCES', - getattr(settings, 'EOX_CORE_USER_ORIGIN_SITE_SOURCES') - ) - return [getattr(cls, source) for source in sources] - - @staticmethod - def fetch_from_created_on_site_prop(user, domain): - """ Fetch option. """ - if not domain: - return False - return UserAttribute.get_user_attribute(user, 'created_on_site') == domain - - @staticmethod - def fetch_from_user_signup_source(user, domain): - """ Read the signup source. """ - return len(UserSignupSource.objects.filter(user=user, site=domain)) > 0 - - @staticmethod - def fetch_from_unfiltered_table(user, site): - """ Fetch option that does not take into account the multi-tentancy model of the installation. """ - return bool(user) - - -def get_course_enrollment(): - """ get CourseEnrollment model """ - return CourseEnrollment - - -def get_user_signup_source(): - """ get UserSignupSource model """ - return UserSignupSource - - -def get_login_failures(): - """ get LoginFailures model """ - return LoginFailures - - -def get_user_profile(): - """ Gets the UserProfile model """ - - return UserProfile - - -def get_user_attribute(): - """ Gets the UserAttribute model """ - return UserAttribute diff --git a/eox_nelp/edxapp_wrapper/backends/users_l_v1_test.py b/eox_nelp/edxapp_wrapper/backends/users_l_v1_test.py deleted file mode 100644 index a7690b42..00000000 --- a/eox_nelp/edxapp_wrapper/backends/users_l_v1_test.py +++ /dev/null @@ -1,102 +0,0 @@ -""" -Test backend to get CourseEnrollment Model. -""" - -from django.contrib.auth.models import Permission - -USERNAME_MAX_LENGTH = 30 - - -def get_edxapp_user(**kwargs): - """ - Return a fake user - """ - return object - - -def create_edxapp_user(*args, **kwargs): - """ - Return a fake user and a list of errors - """ - return object, [] - - -def get_user_read_only_serializer(): - """ - Return a fake user read only serializer - """ - try: - from openedx.core.djangoapps.user_api.accounts.serializers import \ - UserReadOnlySerializer # pylint: disable=import-outside-toplevel - except ImportError: - UserReadOnlySerializer = object - return UserReadOnlySerializer - - -def check_edxapp_account_conflicts(email, username): - """ - Get an executes the check_account_exists for tests - """ - try: - from openedx.core.djangoapps.user_api.accounts.api import \ - check_account_exists # pylint: disable=import-outside-toplevel - except ImportError: - def check_account_exists(email=None, username=None): - """ - Fake definition for check_account_exists edxapp function - """ - return email and username - return check_account_exists(email=email, username=username) - - -def get_course_enrollment(): - """ - Get Test CourseEnrollment model. - - We return any django model that already exists so that - django-filters is happy and no migrations are created. - """ - return Permission - - -def get_course_team_user(*args, **kwargs): - """ - Return a fake course team user - """ - return object - - -def get_user_signup_source(): - """ - Get test UserSignupSource model - """ - try: - from common.djangoapps.student.models import UserSignupSource # pylint: disable=import-outside-toplevel - except ImportError: - UserSignupSource = object - return UserSignupSource - - -def get_user_profile(): - """ Gets the UserProfile model """ - try: - from common.djangoapps.student.models import UserProfile # pylint: disable=import-outside-toplevel - except ImportError: - UserProfile = object - return UserProfile - - -def generate_password(*args, **kwargs): - """ Generates a password """ - return "ThisShouldBeARandomPassword" - - -def get_user_attribute(): - """ - Get test UserAttribute model - """ - try: - from common.djangoapps.student.models import UserAttribute # pylint: disable=import-outside-toplevel - except ImportError: - UserAttribute = object - return UserAttribute diff --git a/eox_nelp/edxapp_wrapper/backends/users_m_v1.py b/eox_nelp/edxapp_wrapper/backends/users_m_v1.py deleted file mode 100644 index 03b4b848..00000000 --- a/eox_nelp/edxapp_wrapper/backends/users_m_v1.py +++ /dev/null @@ -1,357 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -Backend for the create_edxapp_user that works under the open-release/lilac.master tag -""" -import logging - -from common.djangoapps.student.helpers import ( # pylint: disable=import-error,no-name-in-module - create_or_set_user_attribute_created_on_site, - do_create_account, -) -from common.djangoapps.student.models import ( # pylint: disable=import-error,no-name-in-module - CourseEnrollment, - LoginFailures, - Registration, - UserAttribute, - UserProfile, - UserSignupSource, - create_comments_service_user, - email_exists_or_retired, - get_retired_email_by_email, - username_exists_or_retired, -) -from django import forms -from django.conf import settings -from django.contrib.auth import get_user_model -from django.db import transaction -from edx_django_utils.user import generate_password # pylint: disable=import-error,unused-import -from openedx.core.djangoapps.lang_pref import LANGUAGE_KEY # pylint: disable=import-error -from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers # pylint: disable=import-error -from openedx.core.djangoapps.user_api.accounts import USERNAME_MAX_LENGTH # pylint: disable=import-error,unused-import -from openedx.core.djangoapps.user_api.accounts.serializers import UserReadOnlySerializer # pylint: disable=import-error -from openedx.core.djangoapps.user_api.accounts.views import \ - _set_unusable_password # pylint: disable=import-error,unused-import -from openedx.core.djangoapps.user_api.models import UserRetirementStatus # pylint: disable=import-error -from openedx.core.djangoapps.user_api.preferences import api as preferences_api # pylint: disable=import-error -from openedx.core.djangoapps.user_authn.views.register import \ - REGISTER_USER as post_register # pylint: disable=import-error -from openedx.core.djangoapps.user_authn.views.registration_form import ( # pylint: disable=import-error - AccountCreationForm, -) -from openedx.core.djangolib.oauth2_retirement_utils import \ - retire_dot_oauth2_models # pylint: disable=import-error,unused-import -from openedx_events.learning.data import UserData, UserPersonalData # pylint: disable=import-error -from openedx_events.learning.signals import STUDENT_REGISTRATION_COMPLETED # pylint: disable=import-error -from rest_framework import status -from rest_framework.exceptions import NotFound -from social_django.models import UserSocialAuth # pylint: disable=import-error - -LOG = logging.getLogger(__name__) -User = get_user_model() # pylint: disable=invalid-name - - -def get_user_read_only_serializer(): - """ - Great serializer that fits our needs - """ - return UserReadOnlySerializer - - -def check_edxapp_account_conflicts(email, username): - """ - Exposed function to check conflicts - """ - conflicts = [] - - if username and username_exists_or_retired(username): - conflicts.append("username") - - if email and email_exists_or_retired(email): - conflicts.append("email") - - return conflicts - - -class EdnxAccountCreationForm(AccountCreationForm): - """ - A form to extend the behaviour of the AccountCreationForm. - For now the purpose of this form is to allow to make the - password optional if the flag 'skip_password' is True. - This form it's currently only used for validation, not rendering. - """ - - def __init__( # pylint:disable=too-many-arguments - self, - data=None, - extra_fields=None, - extended_profile_fields=None, - do_third_party_auth=True, - tos_required=True, - ): - super().__init__( - data=data, - extra_fields=extra_fields, - extended_profile_fields=extended_profile_fields, - do_third_party_auth=do_third_party_auth, - tos_required=tos_required, - ) - - if data.pop("skip_password", False): - self.fields['password'] = forms.CharField(required=False) - - -def create_edxapp_user(*args, **kwargs): - """ - Creates a user on the open edx django site using calls to - functions defined in the edx-platform codebase - - Example call: - - data = { - 'email': "address@example.org", - 'username': "Username", - 'password': "P4ssW0rd", - 'fullname': "Full Name", - 'activate': True, - 'site': request.site, - 'language_preference': 'es-419', - } - user = create_edxapp_user(**data) - - """ - errors = [] - - extra_fields = getattr(settings, "REGISTRATION_EXTRA_FIELDS", {}) - extended_profile_fields = getattr(settings, "extended_profile_fields", []) - kwargs["name"] = kwargs.pop("fullname", None) - email = kwargs.get("email") - username = kwargs.get("username") - conflicts = check_edxapp_account_conflicts(email=email, username=username) - if conflicts: - return None, [f"Fatal: account collition with the provided: {', '.join(conflicts)}"] - - # Go ahead and create the new user - with transaction.atomic(): - # In theory is possible to extend the registration form with a custom app - # An example form app for this can be found at http://github.com/open-craft/custom-form-app - # form = get_registration_extension_form(data=params) - # if not form: - form = EdnxAccountCreationForm( - data=kwargs, - tos_required=False, - extra_fields=extra_fields, - extended_profile_fields=extended_profile_fields, - # enforce_password_policy=enforce_password_policy, - ) - (user, profile, registration) = do_create_account(form) # pylint: disable=unused-variable - - site = kwargs.pop("site", False) - if site: - create_or_set_user_attribute_created_on_site(user, site) - else: - errors.append("The user was not assigned to any site") - - try: - create_comments_service_user(user) - except Exception: # pylint: disable=broad-except - errors.append("No comments_service_user was created") - - # TODO: link account with third party auth - - # Announce registration through API call - post_register.send_robust(sender=None, user=user) # pylint: disable=no-member - - STUDENT_REGISTRATION_COMPLETED.send_event( - user=UserData( - pii=UserPersonalData( - username=user.username, - email=user.email, - name=user.profile.name, # pylint: disable=no-member - ), - id=user.id, - is_active=user.is_active, - ), - ) - - lang_pref = kwargs.pop("language_preference", False) - if lang_pref: - try: - preferences_api.set_user_preference(user, LANGUAGE_KEY, lang_pref) - except Exception: # pylint: disable=broad-except - errors.append(f"Could not set lang preference '{lang_pref}' for user '{user.username}'") - - if kwargs.pop("activate_user", False): - user.is_active = True - user.save() - - # TODO: run conditional email sequence - - return user, errors - - -def get_edxapp_user(**kwargs): - """ - Retrieve a user by username and/or email - - The user will be returned only if it belongs to the calling site - - Examples: - >>> get_edxapp_user( - { - "username": "Bob", - "site": request.site - } - ) - >>> get_edxapp_user( - { - "email": "Bob@mailserver.com", - "site": request.site - } - ) - """ - params = {key: kwargs.get(key) for key in ['username', 'email'] if key in kwargs} - site = kwargs.get('site') - try: - domain = site.domain - except AttributeError: - domain = None - - try: - user = User.objects.get(**params) - for source_method in FetchUserSiteSources.get_enabled_source_methods(): - if source_method(user, domain): - break - else: - raise User.DoesNotExist - except User.DoesNotExist: - raise NotFound(f'No user found by {str(params)} on site {domain}.') from User.DoesNotExist - return user - - -def delete_edxapp_user(*args, **kwargs): - """ - Deletes a user from the platform. - """ - msg = None - - user = kwargs.get("user") - case_id = kwargs.get("case_id") - site = kwargs.get("site") - is_support_user = kwargs.get("is_support_user") - - user_response = f"The user {user.username} <{user.email}> " - - signup_sources = user.usersignupsource_set.all() - sources = [signup_source.site for signup_source in signup_sources] - - if site and site.name.upper() in (source.upper() for source in sources): - if len(sources) == 1: - with transaction.atomic(): - support_label = "_support" if is_support_user else "" - user.email = f"{user.email}{case_id}.ednx{support_label}_retired" - user.save() - - # Add user to retirement queue. - UserRetirementStatus.create_retirement(user) - - # Unlink LMS social auth accounts - UserSocialAuth.objects.filter(user_id=user.id).delete() - - # Change LMS password & email - user.email = get_retired_email_by_email(user.email) - user.save() - _set_unusable_password(user) - - # Remove the activation keys sent by email to the user for account activation. - Registration.objects.filter(user=user).delete() - - # Delete OAuth tokens associated with the user. - retire_dot_oauth2_models(user) - - # Delete user signup source object - signup_sources[0].delete() - - msg = f"{user_response} has been removed" - else: - for signup_source in signup_sources: - if signup_source.site.upper() == site.name.upper(): - signup_source.delete() - - msg = f"{user_response} has more than one signup source. The signup source from the site {site} has been deleted" - - return msg, status.HTTP_200_OK - - raise NotFound(f"{user_response} does not have a signup source on the site {site}") - - -def get_course_team_user(*args, **kwargs): - """ - Get _course_team_user function. - We need to check if the SERVICE_VARIANT is equal to cms, since - contentstore is a module registered in the INSTALLED_APPS - of the cms only. - """ - if settings.SERVICE_VARIANT == 'cms': - from contentstore.views.user import _course_team_user # pylint: disable=import-error,import-outside-toplevel - return _course_team_user(*args, **kwargs) - return None - - -class FetchUserSiteSources: - """ - Methods to make the comparison to check if an user belongs to a site plus the - get_enabled_source_methods that just brings an array of functions enabled to do so - """ - - @classmethod - def get_enabled_source_methods(cls): - """ Brings the array of methods to check if an user belongs to a site. """ - sources = configuration_helpers.get_value( - 'EOX_CORE_USER_ORIGIN_SITE_SOURCES', - getattr(settings, 'EOX_CORE_USER_ORIGIN_SITE_SOURCES') - ) - return [getattr(cls, source) for source in sources] - - @staticmethod - def fetch_from_created_on_site_prop(user, domain): - """ Fetch option. """ - if not domain: - return False - return UserAttribute.get_user_attribute(user, 'created_on_site') == domain - - @staticmethod - def fetch_from_user_signup_source(user, domain): - """ Read the signup source. """ - return len(UserSignupSource.objects.filter(user=user, site=domain)) > 0 - - @staticmethod - def fetch_from_unfiltered_table(user, site): - """ Fetch option that does not take into account the multi-tentancy model of the installation. """ - return bool(user) - - -def get_course_enrollment(): - """ get CourseEnrollment model """ - return CourseEnrollment - - -def get_user_signup_source(): - """ get UserSignupSource model """ - return UserSignupSource - - -def get_login_failures(): - """ get LoginFailures model """ - return LoginFailures - - -def get_user_profile(): - """ Gets the UserProfile model """ - - return UserProfile - - -def get_user_attribute(): - """ Gets the UserAttribute model """ - return UserAttribute diff --git a/eox_nelp/edxapp_wrapper/bearer_authentication.py b/eox_nelp/edxapp_wrapper/bearer_authentication.py deleted file mode 100644 index 056232c6..00000000 --- a/eox_nelp/edxapp_wrapper/bearer_authentication.py +++ /dev/null @@ -1,17 +0,0 @@ -""" -Authentication definitions. -""" -from importlib import import_module - -from django.conf import settings - - -def get_bearer_authentication(): - """ Gets BearerAuthentication class. """ - backend_function = settings.eox_nelp_BEARER_AUTHENTICATION - backend = import_module(backend_function) - - return backend.get_bearer_authentication() - - -BearerAuthentication = get_bearer_authentication() # pylint: disable=invalid-name diff --git a/eox_nelp/edxapp_wrapper/configuration_helpers.py b/eox_nelp/edxapp_wrapper/configuration_helpers.py deleted file mode 100644 index c75acf59..00000000 --- a/eox_nelp/edxapp_wrapper/configuration_helpers.py +++ /dev/null @@ -1,11 +0,0 @@ -""" Backend abstraction. """ -from importlib import import_module - -from django.conf import settings - - -def get_configuration_helper(*args, **kwargs): - """ Get configuration helper module """ - backend_function = settings.EOX_CORE_CONFIGURATION_HELPER_BACKEND - backend = import_module(backend_function) - return backend.get_configuration_helper(*args, **kwargs) diff --git a/eox_nelp/edxapp_wrapper/storages.py b/eox_nelp/edxapp_wrapper/storages.py deleted file mode 100644 index 330bbffb..00000000 --- a/eox_nelp/edxapp_wrapper/storages.py +++ /dev/null @@ -1,28 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -Storages public function definitions -""" -from importlib import import_module - -from django.conf import settings - - -def get_edxapp_production_staticfiles_storage(): # pylint: disable=invalid-name - """ - Return the edx-platform production staticfiles storage - """ - backend_function = settings.EOX_CORE_STORAGES_BACKEND - backend = import_module(backend_function) - - return backend.get_edxapp_production_staticfiles_storage() - - -def get_edxapp_development_staticfiles_storage(): # pylint: disable=invalid-name - """ - Return the edx-platform production staticfiles storage - """ - backend_function = settings.EOX_CORE_STORAGES_BACKEND - backend = import_module(backend_function) - - return backend.get_edxapp_development_staticfiles_storage() diff --git a/eox_nelp/edxapp_wrapper/third_party_auth.py b/eox_nelp/edxapp_wrapper/third_party_auth.py deleted file mode 100644 index e570686e..00000000 --- a/eox_nelp/edxapp_wrapper/third_party_auth.py +++ /dev/null @@ -1,14 +0,0 @@ -""" -Third Party Auth Exception Middleware definitions. -""" - -from importlib import import_module - -from django.conf import settings - - -def get_tpa_exception_middleware(): - """Get the ExceptionMiddleware class.""" - backend_function = settings.EOX_CORE_THIRD_PARTY_AUTH_BACKEND - backend = import_module(backend_function) - return backend.get_tpa_exception_middleware() diff --git a/eox_nelp/logging.py b/eox_nelp/logging.py deleted file mode 100644 index 5d2d4119..00000000 --- a/eox_nelp/logging.py +++ /dev/null @@ -1,60 +0,0 @@ -""" -The logging module defines functions that are used for the logging of the -pipeline steps. -""" -import logging -import sys -from pprint import pformat - -LOG = logging.getLogger(__name__) - - -# pylint: disable=unused-argument,keyword-arg-before-vararg -def logging_pipeline_step(level, log_message, **local_vars): - """ - This function logs an info or error level message with the information of the pipeline step. - If the logLevel setting is in DEBUG mode then it logs all the arguments passed to the step, if not - it will only show the name of the step, user, backend name, redirect_uri and a custom message. - - For example: - To set the verbosity level to DEBUG add the setting: - - "BACKEND_OPTIONS": { "logLevel":"DEBUG" } - - Arguments: - - level: This is used to describe the severity of the message that the django logger will handle. - If "error" is passed, then an ERROR level logging message will be displayed. Otherwise, - an INFO log will be displayed. - - log_message: Custom message to be shown in the log. - - local_vars: local variables, this must always be passed as **locals(). - """ - extra_info = {} - user = local_vars.get("user") - backend = local_vars.get("backend") - - step = sys._getframe(1).f_code.co_name # pylint: disable=protected-access - backend_name = getattr(backend, "name", "") - uri = getattr(backend, "redirect_uri", "") - msg = log_message - - message = f"PIPELINE-STEP:{step} - USER:{user} - BACKEND:{backend_name} - REDIRECT_URI:{uri} - MESSAGE:{msg}" - - if backend and backend.setting("BACKEND_OPTIONS", {}).get("logLevel", "").lower() == "debug": - details = local_vars.get("details") - args = local_vars.get("args", []) - kwargs = local_vars.get("kwargs", {}) - response = kwargs.pop("response", None) - # We do not want to show the id_token in the logs - if response: - response.pop("id_token", None) - - extra_info["details"] = pformat(details) - extra_info["pipeline_step_args"] = pformat(args) - extra_info["request"] = pformat(kwargs.pop("request", None)) - extra_info["response"] = pformat(response) - extra_info["kwargs"] = pformat(kwargs) - - if level.lower() == "error": - LOG.exception(message, extra=extra_info) - else: - LOG.info(message, extra=extra_info) diff --git a/eox_nelp/middleware.py b/eox_nelp/middleware.py deleted file mode 100644 index cf87976a..00000000 --- a/eox_nelp/middleware.py +++ /dev/null @@ -1,290 +0,0 @@ -#!/usr/bin/python -""" -This file implements the Middleware support for the Open edX platform. -A microsite enables the following features: -1) Mapping of sub-domain name to a 'brand', e.g. foo-university.edx.org -2) Present a landing page with a listing of courses that are specific to the 'brand' -3) Ability to swap out some branding elements in the website -""" -import logging -import re -from urllib.parse import urlparse - -import six -from django.conf import settings -from django.contrib.auth import REDIRECT_FIELD_NAME -from django.contrib.auth.views import redirect_to_login -from django.db import IntegrityError -from django.db.models.signals import post_save -from django.dispatch import receiver -from django.http import Http404, HttpResponseRedirect -from django.urls import reverse -from django.utils.deprecation import MiddlewareMixin -from requests.exceptions import HTTPError - -from eox_nelp.edxapp_wrapper.configuration_helpers import get_configuration_helper -from eox_nelp.edxapp_wrapper.third_party_auth import get_tpa_exception_middleware -from eox_nelp.models import Redirection -from eox_nelp.utils import cache, fasthash - -try: - from social_core.exceptions import AuthAlreadyAssociated, AuthFailed, AuthUnreachableProvider -except ImportError: - AuthUnreachableProvider = Exception - AuthAlreadyAssociated = Exception - AuthFailed = Exception - -try: - from eox_tenant.pipeline import EoxTenantAuthException -except ImportError: - - class EoxTenantAuthException: - """Dummy eox-tenant Exception.""" - - -configuration_helper = get_configuration_helper() # pylint: disable=invalid-name -ExceptionMiddleware = get_tpa_exception_middleware() - -LOG = logging.getLogger(__name__) - - -class PathRedirectionMiddleware(MiddlewareMixin): - """ - Middleware to create custom responses based on the request path - """ - - def process_request(self, request): - """ - Process the path of every request and - determine the correct custom redirect. - - Each custom_redirection_setting is equal in terms of importance, - but to have order, 'MKTG_REDIRECTS' is defined - as the one with highest priority. - """ - - custom_redirection_settings = { - "MKTG_REDIRECTS": "process_mktg_redirect", - "EDNX_CUSTOM_PATH_REDIRECTS": "process_custom_path_redirect", - } - - for key, value in six.iteritems(custom_redirection_settings): - if configuration_helper.has_override_value(key): - action = getattr(self, value) - response = action(request) - if response: - return response - - return None - - def process_custom_path_redirect(self, request): - """ - Redirect the request according to the configured action to take. - """ - redirects = configuration_helper.get_value("EDNX_CUSTOM_PATH_REDIRECTS", {}) - - path = request.path_info - - for regex, values in six.iteritems(redirects): - - if isinstance(values, dict): - key = next(iter(values)) - else: - key = values - - regex_path_match = re.compile(regex.format( - COURSE_ID_PATTERN=settings.COURSE_ID_PATTERN, - USERNAME_PATTERN=settings.USERNAME_PATTERN, - )) - - if regex_path_match.match(path): - try: - action = getattr(self, key) - return action(request=request, key=key, values=values, path=path) - except Http404: # we expect 404 to be raised - raise - except Exception as error: # pylint: disable=broad-except - LOG.error("The PathRedirectionMiddleware generated an error at: %s%s", - request.get_host(), - request.get_full_path()) - LOG.error(error) - return None - return None - - def process_mktg_redirect(self, request): - """ - Redirect the request if there are mktg custom settings - present that match the request path. - """ - redirects = configuration_helper.get_value("MKTG_REDIRECTS", {}) - path = request.path_info - - for key, value in six.iteritems(redirects): - - # Strip off html extension to have backwards - # compatibility to keys defined with template style. - key = key.replace('.html', '') - key = f'/{key}' - - # Just continue if the path does not match or the redirect value is empty - # TODO: validate that the key corresponds to a Marketing path - if path != key or not value: - continue - - try: - values = {key: value} - return self.redirect_always(key=key, values=values) - except Exception as error: # pylint: disable=broad-except - LOG.error("The PathRedirectionMiddleware generated an error at: %s%s", - request.get_host(), - request.get_full_path()) - LOG.error(error) - return None - return None - - def login_required(self, request, path, **kwargs): # pylint: disable=unused-argument - """ - Action: a user session must exist. - If it does not, redirect to the login page - """ - if request.user.is_authenticated: - return None - resolved_login_url = configuration_helper.get_dict("FEATURES", {}).get( - "ednx_custom_login_link", settings.LOGIN_URL) - return redirect_to_login(path, resolved_login_url, REDIRECT_FIELD_NAME) - - def not_found(self, **kwargs): # pylint: disable=unused-argument - """ - Action: return 404 error for anyone - """ - raise Http404 - - def not_found_loggedin(self, request, **kwargs): # pylint: disable=unused-argument - """ - Action: return 404 error for users which have a session - """ - if request.user.is_authenticated: - raise Http404 - - def not_found_loggedout(self, request, **kwargs): # pylint: disable=unused-argument - """ - Action: return 404 error for users that don't have a session - """ - if not request.user.is_authenticated: - raise Http404 - - def redirect_always(self, key, values, **kwargs): # pylint: disable=unused-argument - """ - Action: redirect to the given target - """ - return HttpResponseRedirect(values[key]) - - def redirect_loggedin(self, request, key, values, **kwargs): # pylint: disable=unused-argument - """ - Action: redirect logged users to the given target - """ - if request.user.is_authenticated: - return HttpResponseRedirect(values[key]) - return None - - def redirect_loggedout(self, request, key, values, **kwargs): # pylint: disable=unused-argument - """ - Action: redirect external visitors to the given target - """ - if not request.user.is_authenticated: - return HttpResponseRedirect(values[key]) - return None - - -class RedirectionsMiddleware(MiddlewareMixin): - """ - Middleware for Redirecting microsites to other domains or to error pages - """ - - def process_request(self, request): - """ - This middleware handles redirections and error pages according to the - business logic at edunext - """ - if not settings.FEATURES.get('USE_REDIRECTION_MIDDLEWARE', True): - return None - - domain = request.META.get('HTTP_HOST', "") - - # First handle the event where a domain has a redirect target - cache_key = "redirect_cache." + fasthash(domain) - target = cache.get(cache_key) # pylint: disable=maybe-no-member - - if not target: - try: - target = Redirection.objects.get(domain__iexact=domain) # pylint: disable=no-member - except Redirection.DoesNotExist: # pylint: disable=no-member - target = '##none' - - cache.set( # pylint: disable=maybe-no-member - cache_key, target, 5 * 60 - ) - - if target != '##none': - # If we are already at the target, just return - if domain == target.target and request.scheme == target.scheme: # pylint: disable=no-member - return None - - to_url = f'{target.scheme}://{target.target}{request.path}' - - return HttpResponseRedirect( - to_url, - status=target.status, # pylint: disable=no-member - ) - return None - - @staticmethod - @receiver(post_save, sender=Redirection) - def clear_cache(sender, instance, **kwargs): # pylint: disable=unused-argument - """ - Clear the cached template when the model is saved - """ - cache_key = "redirect_cache." + fasthash(instance.domain) - cache.delete(cache_key) # pylint: disable=maybe-no-member - - -class TPAExceptionMiddleware(ExceptionMiddleware): - """Middleware to handle exceptions not catched by Social Django""" - - def process_exception(self, request, exception): - """ - Handle exceptions raised during the authentication process. - """ - referer_url = request.META.get('HTTP_REFERER', '') - referer_url = urlparse(referer_url).path - - if referer_url != reverse('signin_user') and request.view_name not in ['auth', 'complete']: - return super().process_exception(request, exception) - - if isinstance(exception, EoxTenantAuthException): - new_exception = AuthFailed( - exception.backend, - str(exception), - ) - LOG.error("%s", exception) - return super().process_exception(request, new_exception) - - if isinstance(exception, IntegrityError): - backend = getattr(request, 'backend', None) - new_exception = AuthAlreadyAssociated( - backend, - "The given email address is associated with another account", - ) - LOG.error("%s", exception) - return super().process_exception(request, new_exception) - - if isinstance(exception, HTTPError): - backend = getattr(request, 'backend', None) - new_exception = AuthUnreachableProvider( - backend, - "Unable to connect with the external provider", - ) - LOG.error("%s", exception) - return super().process_exception(request, new_exception) - - return super().process_exception(request, exception) \ No newline at end of file diff --git a/eox_nelp/pipeline.py b/eox_nelp/pipeline.py deleted file mode 100644 index a323337b..00000000 --- a/eox_nelp/pipeline.py +++ /dev/null @@ -1,221 +0,0 @@ -""" -The pipeline module defines functions that are used in the third party authentication flow -""" -import logging - -from crum import get_current_request -from django.db.models.signals import post_save - -from eox_nelp.edxapp_wrapper.users import ( - generate_password, - get_user_attribute, - get_user_profile, - get_user_signup_source, -) -from eox_nelp.logging import logging_pipeline_step - -try: - from social_core.exceptions import AuthFailed, NotAllowedToDisconnect -except ImportError: - AuthFailed = object - NotAllowedToDisconnect = object - -UserSignupSource = get_user_signup_source() # pylint: disable=invalid-name -LOG = logging.getLogger(__name__) - - -# pylint: disable=unused-argument,keyword-arg-before-vararg -def ensure_new_user_has_usable_password(backend, user=None, **kwargs): - """ - This pipeline function assigns an usable password to an user in case that the user has an unusable password on user cretion. - At the creation of new users through some TPA providers, some of them are created with an unusable password, a user with an unusable password cannot login - properly in the platform if the common.djangoapps.third_party.pipeline.set_logged_in_cookies step is enabled. - - See: https://github.com/eduNEXT/edunext-platform/blob/c83b82a4aba2a5496e3b7da83972a8edf25fcdd2/common/djangoapps/third_party_auth/pipeline.py#L673 - - It's recommended to place this step after the social core step that creates the users: (social_core.pipeline.user.create_user). - """ - is_new = kwargs.get('is_new') - - if user and is_new and not user.has_usable_password(): - user.set_password(generate_password(length=25)) - user.save() - - user_attribute_model = get_user_attribute() - user_attribute_model.set_user_attribute(user, 'auto_password_via_tpa_pipeline', 'true') - - logging_pipeline_step( - "info", - "Assigned an usable password to the user on creation.", - **locals() - ) - - -# pylint: disable=unused-argument,keyword-arg-before-vararg -def ensure_user_has_profile(backend, details, user=None, *args, **kwargs): - """ - This pipeline function creates an empty profile object if the user does not have one. - It can be used with the user_details_force_sync function to fill the profile after creation. - """ - if user: - user_profile_model = get_user_profile() - try: - __ = user.profile - except user_profile_model.DoesNotExist: - user_profile_model.objects.create(user=user) - logging_pipeline_step( - "info", - "Created new profile for user during the third party pipeline.", - **locals() - ) - - -def force_user_post_save_callback(user=None, *args, **kwargs): - """ - Send the signal post_save in order to force the execution of user_post_save_callback, see it in - https://github.com/eduNEXT/edunext-platform/blob/4169327231de00991c46b6192327fe50b0406561/common/djangoapps/student/models.py#L652 - this allows the automatic enrollments if a user is registered by a third party auth. - - It's recommended to place this step after the social core step that creates the users: (social_core.pipeline.user.create_user). - - the discussion left three options: - - 1) - user._changed_fields = {'is_active': user.is_active} # pylint: disable=protected-access - post_save.send_robust( - sender=user.__class__, - instance=user, - created=False - ) - - 2) - user.is_active = False - user.save() - user.is_active = True - user.save() - - 3) - registration = Registration.objects.get_or_create(user=user) - user.is_active = False - user.save() - registration.activate() - - The first one was selected because the second executed multiple unnecessary process and third one - needs to implement a new backend. - """ - is_new = kwargs.get('is_new') - - if user and is_new: - user._changed_fields = {'is_active': user.is_active} # pylint: disable=protected-access - post_save.send_robust( - sender=user.__class__, - instance=user, - created=False - ) - logging_pipeline_step( - "info", - "Post_save signal sent in order to force the execution of user_post_save_callback.", - **locals() - ) - - -def check_disconnect_pipeline_enabled(backend, *args, **kwargs): - """ - This pipeline function checks whether disconnection from the auth provider is enabled or not. That's - done checking for `disableDisconnectPipeline` setting defined in the provider configuration. - - For example: - To avoid disconnection from SAML, add the following to `Other config str` in your SAMLConfiguration: - - "BACKEND_OPTIONS": { "disableDisconnectPipeline":true }, - - Now, to avoid disconnection from an Oauth2Provider, add the same setting to `Other settings` in your - Oauth2Provider. - - It's recommended to place this function at the beginning of the pipeline. - """ - if backend and backend.setting("BACKEND_OPTIONS", {}).get("disableDisconnectPipeline"): - logging_pipeline_step( - "error", - "Disconnection pipeline is disabled, users are not allowed to disconnect.", - **locals() - ) - raise NotAllowedToDisconnect() # pylint: disable=raising-non-exception - - -def assert_user_information(details, user, backend, *args, **kwargs): - """ - This pipeline function checks whether the user information from the LMS matches the information - returned by the provider. - - For example: - To avoid connection when the user's email from the LMS differs from the email returned by the provider, - add the setting: - - "BACKEND_OPTIONS": { "assertUserInformationMatchFields":["email",...] } - - It's recommended to place this step after the user is available (ie. after ensure_user_information step) and - before user association (ie. before user_association step) - """ - if not (user and backend): - return - - defined_fields = backend.setting("BACKEND_OPTIONS", {}).get("assertUserInformationMatchFields", []) - - for field in defined_fields: - - if str(details.get(field, "")) != str(getattr(user, field, "")): - logging_pipeline_step( - "error", - f"Credentials not allowed: field {field} returned by provider does not match with the users information.", - **locals() - ) - raise AuthFailed(backend, "Credentials not allowed.") # pylint: disable=raising-non-exception - - -def create_signup_source_for_new_association(user=None, *args, **kwargs): - """ - This pipeline step registers a new signup source for users with new social auth link. The signup source - will have associated the current site and user. - - It's recommended to place this step after the creation of the social auth link, i.e, after the step - 'social_core.pipeline.social_auth.associate_user', - - To avoid errors or unnecessary searches, these steps are skipped using 'is_new' key insterted by - 'social_core.pipeline.user.create_user' after creating a user. - """ - if not user or kwargs.get("is_new", False): - return - - if kwargs.get("new_association", False): - _, created = UserSignupSource.objects.get_or_create(user=user, site=get_current_request().site) - if created: - logging_pipeline_step( - "info", - "Created new singup source for user during the third party pipeline.", - **locals() - ) - - -def ensure_user_has_signup_source(user=None, *args, **kwargs): - """ - This pipeline step registers a new signup source for users that have a social auth link but - no signup source associated with the current site. The signup source will be associated to - the current site and user. - - It's recommended to place this step at the end of the TPA pipeline, e.g., after the step - 'eox_nelp.pipeline.ensure_user_has_profile', - - To avoid errors or unnecessary searches, these steps are skipped using 'is_new' key insterted by - 'social_core.pipeline.user.create_user' after creating a user. - """ - if not user or kwargs.get("is_new", False): - return - - _, created = UserSignupSource.objects.get_or_create(user=user, site=get_current_request().site) - if created: - logging_pipeline_step( - "info", - "Created new singup source for user during the third party pipeline.", - **locals() - ) diff --git a/eox_nelp/settings/aws.py b/eox_nelp/settings/aws.py deleted file mode 100644 index 996ab32c..00000000 --- a/eox_nelp/settings/aws.py +++ /dev/null @@ -1,9 +0,0 @@ -""" -AWS Django settings for eox_nelp project. - -Juniper release will remove aws.py file. https://openedx.atlassian.net/browse/DEPR-14 -""" - -from __future__ import unicode_literals - -from .production import plugin_settings # pylint: disable=unused-import diff --git a/eox_nelp/settings/devstack.py b/eox_nelp/settings/devstack.py index fdd98336..b84bd301 100644 --- a/eox_nelp/settings/devstack.py +++ b/eox_nelp/settings/devstack.py @@ -9,6 +9,3 @@ def plugin_settings(settings): # pylint: disable=function-redefined Set of plugin settings used by the Open Edx platform. More info: https://github.com/edx/edx-platform/blob/master/openedx/core/djangoapps/plugins/README.rst """ - settings.eox_nelp_STATICFILES_STORAGE = "eox_nelp.storage.DevelopmentStorage" - if settings.eox_nelp_ENABLE_STATICFILES_STORAGE: - settings.STATICFILES_STORAGE = settings.eox_nelp_STATICFILES_STORAGE \ No newline at end of file diff --git a/eox_nelp/settings/production.py b/eox_nelp/settings/production.py index a0736567..faf8c049 100644 --- a/eox_nelp/settings/production.py +++ b/eox_nelp/settings/production.py @@ -15,74 +15,3 @@ def plugin_settings(settings): # pylint: disable=function-redefined Set of plugin settings used by the Open Edx platform. More info: https://github.com/edx/edx-platform/blob/master/openedx/core/djangoapps/plugins/README.rst """ - - settings.eox_nelp_BEARER_AUTHENTICATION = getattr(settings, 'ENV_TOKENS', {}).get( - 'eox_nelp_BEARER_AUTHENTICATION', - settings.eox_nelp_BEARER_AUTHENTICATION - ) - - - - - - - - settings.eox_nelp_THIRD_PARTY_AUTH_BACKEND = getattr(settings, 'ENV_TOKENS', {}).get( - 'eox_nelp_THIRD_PARTY_AUTH_BACKEND', - settings.eox_nelp_THIRD_PARTY_AUTH_BACKEND - ) - settings.eox_nelp_USER_ENABLE_MULTI_TENANCY = getattr(settings, 'ENV_TOKENS', {}).get( - 'eox_nelp_USER_ENABLE_MULTI_TENANCY', - settings.eox_nelp_USER_ENABLE_MULTI_TENANCY - ) - if not settings.eox_nelp_USER_ENABLE_MULTI_TENANCY: - user_origin_sources = [ - 'fetch_from_unfiltered_table', - ] - else: - user_origin_sources = settings.eox_nelp_USER_ORIGIN_SITE_SOURCES - settings.eox_nelp_USER_ORIGIN_SITE_SOURCES = getattr(settings, 'ENV_TOKENS', {}).get( - 'eox_nelp_USER_ORIGIN_SITE_SOURCES', - user_origin_sources - ) - - settings.eox_nelp_APPEND_LMS_MIDDLEWARE_CLASSES = getattr(settings, 'ENV_TOKENS', {}).get( - 'eox_nelp_APPEND_LMS_MIDDLEWARE_CLASSES', - settings.eox_nelp_APPEND_LMS_MIDDLEWARE_CLASSES - ) - if settings.SERVICE_VARIANT == "lms": - if settings.eox_nelp_APPEND_LMS_MIDDLEWARE_CLASSES: - settings.MIDDLEWARE += [ - 'eox_nelp.middleware.PathRedirectionMiddleware', - 'eox_nelp.middleware.RedirectionsMiddleware', - 'eox_nelp.middleware.TPAExceptionMiddleware' - ] - - # Sentry Integration - sentry_integration_dsn = getattr(settings, 'ENV_TOKENS', {}).get( - 'eox_nelp_SENTRY_INTEGRATION_DSN', - settings.eox_nelp_SENTRY_INTEGRATION_DSN - ) - settings.eox_nelp_SENTRY_IGNORED_ERRORS = getattr(settings, 'ENV_TOKENS', {}).get( - 'eox_nelp_SENTRY_IGNORED_ERRORS', - settings.eox_nelp_SENTRY_IGNORED_ERRORS - ) - sentry_environment = getattr(settings, 'ENV_TOKENS', {}).get( - 'eox_nelp_SENTRY_ENVIRONMENT', - settings.eox_nelp_SENTRY_ENVIRONMENT - ) - - if sentry_sdk is not None and sentry_integration_dsn is not None: - from eox_nelp.integrations.sentry import ExceptionFilterSentry # pylint: disable=import-outside-toplevel - sentry_sdk.init( - before_send=ExceptionFilterSentry(), - dsn=sentry_integration_dsn, - environment=sentry_environment, - integrations=[ - DjangoIntegration(), - ], - - # If you wish to associate users to errors (assuming you are using - # django.contrib.auth) you may enable sending PII data. - send_default_pii=True - ) diff --git a/eox_nelp/social_tpa_backends.py b/eox_nelp/social_tpa_backends.py deleted file mode 100644 index e69de29b..00000000 diff --git a/eox_nelp/utils.py b/eox_nelp/utils.py deleted file mode 100644 index edecfda7..00000000 --- a/eox_nelp/utils.py +++ /dev/null @@ -1,121 +0,0 @@ -#!/usr/bin/python -""" -Util function definitions. -""" -import datetime -import hashlib - -from django.conf import settings -from django.core import cache -from pytz import UTC -from rest_framework import serializers - - -UserProfile = get_user_profile() - -try: - cache = cache.caches['general'] # pylint: disable=invalid-name -except Exception: # pylint: disable=broad-except - cache = cache.cache # pylint: disable=invalid-name - - -def fasthash(string): - """ - Hashes `string` into a string representation of a 128-bit digest. - """ - md4 = hashlib.new("md4") - md4.update(string.encode('utf-8')) - return md4.hexdigest() - - -def get_valid_years(): - """ - Return valid list of year range, for the YEAR_OF_BIRTH_CHOICES - constant. - """ - current_year = datetime.datetime.now(UTC).year - return list(range(current_year, current_year - 120, -1)) - - -def get_gender_choices(): - """ - Try to return the valid options for the UserProfile field "gender" - """ - return getattr(UserProfile, "GENDER_CHOICES", ()) - - -def get_level_of_education_choices(): - """ - Try to return the valid options for the UserProfile field "level_of_education". - """ - return getattr(UserProfile, "LEVEL_OF_EDUCATION_CHOICES", ()) - - -def set_custom_field_restrictions(custom_field, serializer_field): - """ - Given a custom_field definition dict, check for any - restriction and add it to the serializer field dictionary - - This function is only called for custom fields of type - "text". The allowed restrictions are taken from the platform's - ALLOWED_RESTRICTIONS dictionary - - ALLOWED_RESTRICTIONS = { - "text": ["min_length", "max_length"], - "password": ["min_length", "max_length", "min_upper", "min_lower", - "min_punctuation", "min_symbol", "min_numeric", "min_alphabetic"], - "email": ["min_length", "max_length", "readonly"], - } - """ - for key, value in custom_field.get("restrictions", {}).items(): - try: - ["min_length", "max_length"].index(key) - serializer_field[key] = int(value) - except ValueError as err: - raise serializers.ValidationError({"restriction error": f"{key}: {value}.\ - The restriction may not be valid or the value is not an integer"}) from err - - return serializer_field - - -def set_select_custom_field(custom_field, serializer_field): - """ - Given a custom_field definition dict with type=='select', - add its options and default value (if exists) to the serializer - field dictionary. - """ - field_name = custom_field.get("name") - choices = custom_field.get("options", []) - serializer_field["choices"] = choices - default = custom_field.get("default") - - if default: - try: - choices.index(default) - serializer_field["default"] = default - # A field can not be both `required` and have a `default` - serializer_field["required"] = False - except ValueError as err: - raise serializers.ValidationError({f"{field_name}": "The default value must be one of the options"}) from err - - return serializer_field - - -def get_registration_extra_fields(): - """ - Return only the registration extra fields - that are not hidden. - These fields are the ones that will be taken into - account to initialize the EdxappExtendedUserSerializer - """ - registration_extra_fields = getattr(settings, 'REGISTRATION_EXTRA_FIELDS', {}) - - return {key: value for key, value in registration_extra_fields.items() if value in ["required", "optional"]} - - -def create_user_profile(user): - """ - Creates a Profile to a User. - """ - if not hasattr(user, "profile"): - UserProfile.objects.create(user=user)