Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

override enrollment attributes for learners

Browse Source 
ENT-2998
This commit is contained in:
muhammad-ammar
2020-06-22 17:22:56 +05:00
parent 0d0f8a1c6f
commit 4a3eb16296
5 changed files with 270 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
openedx/features/enterprise_support/admin/__init__.py Normal file
View File

21
openedx/features/enterprise_support/admin/forms.py Normal file
View File

@@ -0,0 +1,21 @@
"""
Enterprise support admin forms.
"""
from django import forms
from django.utils.translation import ugettext_lazy as _
from enterprise.admin.utils import validate_csv
class CSVImportForm(forms.Form):
csv_file = forms.FileField(
required=True,
label=_('CSV File'),
help_text=_('CSV file should have 3 columns having names lms_user_id, course_id, opportunity_id')
)
def clean_csv_file(self):
csv_file = self.cleaned_data['csv_file']
csv_reader = validate_csv(csv_file, expected_columns=['lms_user_id', 'course_id', 'opportunity_id'])
return csv_reader

71
openedx/features/enterprise_support/admin/views.py Normal file
View File

@@ -0,0 +1,71 @@
"""
Enterprise support admin views.
"""
from django.contrib import admin, messages
from django.urls import reverse
from django.utils.translation import ugettext as _
from django.views.generic.edit import FormView
from enterprise.models import EnterpriseCourseEnrollment
from openedx.features.enterprise_support.admin.forms import CSVImportForm
from student.models import CourseEnrollment, CourseEnrollmentAttribute
class EnrollmentAttributeOverrideView(FormView):
"""
Learner Enrollment Attribute Override View.
"""
template_name = 'enterprise_support/admin/enrollment_attributes_override.html'
form_class = CSVImportForm
@staticmethod
def _get_admin_context(request):
admin_context = {'opts': EnterpriseCourseEnrollment._meta}
return admin_context
def get_success_url(self):
return reverse('admin:enterprise_override_attributes')
def get_context_data(self, **kwargs):
context = super(EnrollmentAttributeOverrideView, self).get_context_data(**kwargs)
context.update(self._get_admin_context(self.request))
return context
def form_valid(self, form):
total_records = 0
error_line_numbers = []
csv_reader = form.cleaned_data['csv_file']
for index, record in enumerate(csv_reader):
total_records += 1
try:
course_enrollment = CourseEnrollment.objects.get(
user_id=record['lms_user_id'],
course_id=record['course_id'],
)
except CourseEnrollment.DoesNotExist:
error_line_numbers.append(str(index + 1))
else:
CourseEnrollmentAttribute.objects.update_or_create(
enrollment=course_enrollment,
namespace='salesforce',
name='opportunity_id',
defaults={
'value': record['opportunity_id'],
}
)
# if for some reason not a single enrollment updated than do not show success message.
if len(error_line_numbers) != total_records:
messages.success(self.request, 'Successfully updated learner enrollment opportunity ids.')
if error_line_numbers:
messages.error(
self.request,
_(
'Enrollment attributes were not updated for records at following line numbers '
'in csv because no enrollment found for these records: {error_line_numbers}'
).format(error_line_numbers=', '.join(error_line_numbers))
)
return super(EnrollmentAttributeOverrideView, self).form_valid(form)

33
openedx/features/enterprise_support/templates/enterprise_support/admin/enrollment_attributes_override.html Normal file
View File

@@ -0,0 +1,33 @@
{% extends "admin/base_site.html" %}
{% load i18n static admin_urls %}
{% block extrastyle %}
<link rel="stylesheet" type="text/css" href="{% static 'enterprise/bundles/main-admin.style.css' %}"/>
<link rel="stylesheet" type="text/css" href="{% static 'admin/css/changelists.css' %}"/>
{% endblock %}
{% block extrahead %}
<script type="text/javascript" src="{% url 'admin:jsi18n' %}"></script>
{% endblock %}
{% block breadcrumbs %}
<div class="breadcrumbs">
<a href="{% url 'admin:index' %}">{% trans "Home" as tmsg %}{{ tmsg | force_escape }}</a>
&rsaquo; <a href="{% url 'admin:app_list' app_label=opts.app_label %}">{{ opts.app_config.verbose_name }}</a>
&rsaquo; <a href="{% url opts|admin_urlname:'changelist' %}">{% trans "Enterprise Course Enrollments" as tmsg %}{{ tmsg | force_escape }}</a>
&rsaquo; {% trans "Override Enrollment Attributes" as tmsg %}{{ tmsg | force_escape }}
</div>
{% endblock %}
{% block content %}
<div id="content-main">
<div class="forms-panel">
<h1>{% trans "Upload CSV to override enrollment attributes for learners" as tmsg %}{{ tmsg | force_escape }}</h1>
<form action="" method="post" enctype="multipart/form-data" id="enrollment-attribute-update-form">
{% csrf_token %}
{{ form.as_p }}
<input type="submit" value="Submit"/>
</form>
</div>
</div>
{% endblock %}

145
openedx/features/enterprise_support/tests/test_admin.py Normal file
View File

@@ -0,0 +1,145 @@
"""
Enterprise support admin tests.
"""
import csv
import os
import tempfile
from django.contrib.messages import get_messages
from django.test import Client
from django.urls import reverse
from openedx.core.djangoapps.catalog.tests.factories import CourseRunFactory
from openedx.features.enterprise_support.admin.forms import CSVImportForm
from student.models import CourseEnrollment, CourseEnrollmentAttribute
from student.tests.factories import TEST_PASSWORD, AdminFactory, CourseEnrollmentFactory, UserFactory
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
class EnrollmentAttributeOverrideViewTest(ModuleStoreTestCase):
"""
Tests for course creator admin.
"""
def setUp(self):
""" Test case setup """
super(EnrollmentAttributeOverrideViewTest, self).setUp()
self.client = Client()
user = AdminFactory()
self.view_url = reverse('admin:enterprise_override_attributes')
self.client.login(username=user.username, password=TEST_PASSWORD)
self.users = []
for _ in range(3):
self.users.append(UserFactory())
self.course = CourseRunFactory()
self.course_id = self.course.get('key')
self.csv_data = [
[self.users[0].id, self.course_id, 'OP_4321'],
[self.users[1].id, self.course_id, 'OP_8765'],
[self.users[2].id, self.course_id, 'OP_2109'],
]
self.csv_data_for_existing_attributes = [
[self.users[0].id, self.course_id, 'OP_1234'],
[self.users[1].id, self.course_id, 'OP_5678'],
[self.users[2].id, self.course_id, 'OP_9012'],
]
for user in self.users:
CourseEnrollmentFactory(
course_id=self.course_id,
user=user
)
def create_csv(self, header=None, data=None):
"""Create csv"""
header = header or ['lms_user_id', 'course_id', 'opportunity_id']
data = data or self.csv_data
tmp_csv_path = os.path.join(tempfile.gettempdir(), 'data.csv')
with open(tmp_csv_path, 'w') as csv_file:
csv_writer = csv.writer(csv_file)
csv_writer.writerow(header)
csv_writer.writerows(data)
return tmp_csv_path
def verify_enrollment_attributes(self, data=None):
"""
Verify that data from csv is imported correctly and tables have correct data.
"""
data = data or self.csv_data
for user_id, course_id, opportunity_id in data:
enrollment = CourseEnrollment.objects.get(user_id=user_id, course_id=course_id)
enrollment_attribute = CourseEnrollmentAttribute.objects.get(
enrollment=enrollment,
namespace='salesforce',
name='opportunity_id'
)
assert enrollment_attribute.value == opportunity_id
def test_get(self):
"""
Tests that HTTP GET is working as expected.
"""
response = self.client.get(self.view_url)
assert response.status_code == 200
assert isinstance(response.context['form'], CSVImportForm)
def test_post(self):
"""
Tests that HTTP POST is working as expected when creating new attributes and updating.
"""
csv_path = self.create_csv()
post_data = {'csv_file': open(csv_path)}
response = self.client.post(self.view_url, data=post_data)
assert response.status_code == 302
self.verify_enrollment_attributes()
# override existing
csv_path = self.create_csv(data=self.csv_data_for_existing_attributes)
post_data = {'csv_file': open(csv_path)}
response = self.client.post(self.view_url, data=post_data)
assert response.status_code == 302
self.verify_enrollment_attributes(data=self.csv_data_for_existing_attributes)
def test_post_with_no_csv(self):
"""
Tests that HTTP POST without out csv file is working as expected.
"""
response = self.client.post(self.view_url)
assert response.context['form'].errors == {'csv_file': ['This field is required.']}
def test_post_with_incorrect_csv_header(self):
"""
Tests that HTTP POST with incorrect csv header is working as expected.
"""
csv_path = self.create_csv(header=['a', 'b'])
post_data = {'csv_file': open(csv_path)}
response = self.client.post(self.view_url, data=post_data)
assert response.context['form'].errors == {
'csv_file': [
'Expected a CSV file with [lms_user_id, course_id, opportunity_id] '
'columns, but found [a, b] columns instead.'
]
}
def test_post_with_no_enrollment_error(self):
"""
Tests that HTTP POST is working as expected when for some records there is no enrollment.
"""
csv_data = self.csv_data + [[999, self.course_id, 'NOPE'], [1000, self.course_id, 'NONE']]
csv_path = self.create_csv(data=csv_data)
post_data = {'csv_file': open(csv_path)}
response = self.client.post(self.view_url, data=post_data)
assert response.status_code == 302
messages = []
for msg in get_messages(response.wsgi_request):
messages.append(str(msg))
assert messages == [
'Successfully updated learner enrollment opportunity ids.',
'Enrollment attributes were not updated for records at following line numbers '
'in csv because no enrollment found for these records: 4, 5'
]