Skip to content
This repository has been archived by the owner on Sep 11, 2024. It is now read-only.

Commit

Permalink
Implement feature and add unit test cases
Browse files Browse the repository at this point in the history
  • Loading branch information
@how015
root authored and how015 committed Apr 9, 2020
1 parent a4c32d7 commit a3783d1
Show file tree
Hide file tree
Showing 2 changed files with 167 additions and 4 deletions.
17 changes: 15 additions & 2 deletions registrar/apps/api/v1/mixins.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import uuid
from collections.abc import Iterable

import waffle
from django.core.exceptions import ImproperlyConfigured, PermissionDenied
from django.http import Http404
from django.utils.functional import cached_property
Expand Down Expand Up @@ -272,7 +273,7 @@ def handle_enrollments(self, course_id=None):
Does program enrollments if `course_id` is None.
Does course run enrollments otherwise.
"""
self.validate_enrollment_data(self.request.data)
self.validate_enrollment_data(self.request, course_id)
if course_id:
good, bad, results = write_course_run_enrollments(
self.request.method,
Expand All @@ -295,10 +296,11 @@ def handle_enrollments(self, course_id=None):
self.add_tracking_data(failure='unprocessable_entity')
return Response(results, status=status)

def validate_enrollment_data(self, enrollments):
def validate_enrollment_data(self, request, course_id=None):
"""
Validate enrollments request body
"""
enrollments = request.data
if not isinstance(enrollments, list):
self.add_tracking_data(failure='bad_request')
raise ValidationError('expected request body type: List')
Expand All @@ -323,3 +325,14 @@ def validate_enrollment_data(self, enrollments):
raise ValidationError(
'expected request dicts to have string value for "status"'
)
if enrollment.get('course_staff') is not None:
if not isinstance(enrollment.get('course_staff'), bool):
self.add_tracking_data(failure='bad_request')
raise ValidationError(
'expected request dicts to have boolean value for "course_staff"'
)
if course_id:
if not waffle.flag_is_active(request, 'enable_course_role_management'):
raise PermissionDenied('"course_staff" not accepted since role assignment is not enabled')
else:
raise PermissionDenied('"course_staff" field is for course only')
154 changes: 152 additions & 2 deletions registrar/apps/api/v1/tests/test_views.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import json
import logging
import uuid
from contextlib import contextmanager
from io import StringIO
from posixpath import join as urljoin

Expand All @@ -21,6 +22,8 @@
from rest_framework.test import APITestCase
from user_tasks.models import UserTaskStatus
from user_tasks.tasks import UserTask
from waffle import get_waffle_flag_model
from waffle.testutils import override_flag

from registrar.apps.api.constants import ENROLLMENT_WRITE_MAX_SIZE
from registrar.apps.api.tests.mixins import AuthRequestMixin, TrackTestMixin
Expand Down Expand Up @@ -69,6 +72,18 @@
ACTIVE_CURRICULUM_UUID = '77777777-4444-2222-1111-000000000000'
INACTIVE_CURRICULUM_UUID = '66666666-4444-2222-1111-000000000000'

@contextmanager
def activate_waffle_flag(flag_name, group):
"""
Activate the given flag on the given group.
"""
waffle_model = get_waffle_flag_model()
waffle_flag = waffle_model.objects.create(name=flag_name)
waffle_flag.groups.add(group)
waffle_flag.save()
waffle_flag.flush()
yield waffle_flag
waffle_flag.delete()

class RegistrarAPITestCase(TrackTestMixin, APITestCase):
""" Base for tests of the Registrar API """
Expand Down Expand Up @@ -164,6 +179,14 @@ def setUpTestData(cls):
)
cls.program_user.groups.add(cls.program_group) # pylint: disable=no-member

cls.cs_program_admin = UserFactory(username='cs-program-admin')
cls.cs_program_admin_group = ProgramOrganizationGroupFactory(
name='cs-program-admins',
program=cls.cs_program,
role=perms.ProgramReadWriteEnrollmentsRole.name
)
cls.cs_program_admin.groups.add(cls.cs_program_admin_group) # pylint: disable=no-member

def setUp(self):
super().setUp()
self._add_programs_to_cache()
Expand Down Expand Up @@ -1497,10 +1520,11 @@ def get_url(self, program_key=None, course_id=None):
def mock_course_enrollments_response(self, method, expected_response, response_code=200):
self.mock_api_response(self.lms_request_url, expected_response, method=method, response_code=response_code)

def student_course_enrollment(self, status, student_key=None):
def student_course_enrollment(self, status, student_key=None, course_staff=None):
return {
'status': status,
'student_key': student_key or uuid.uuid4().hex[0:10]
'student_key': student_key or uuid.uuid4().hex[0:10],
'course_staff': course_staff
}

def test_program_unauthorized_at_organization(self):
Expand Down Expand Up @@ -1618,19 +1642,145 @@ def test_successful_program_course_enrollment_write(self, use_external_course_ke
{
'status': 'active',
'student_key': '001',
'course_staff': None,
},
{
'status': 'active',
'student_key': '002',
'course_staff': None,
},
{
'status': 'inactive',
'student_key': '003',
'course_staff': None,
}
])
self.assertEqual(response.status_code, 200)
self.assertDictEqual(response.data, expected_lms_response)

@mock_oauth_login
@responses.activate
@ddt.data(False, True)
def test_successful_update_course_staff_with_organization_group_waffle(self, use_external_course_key):
course_id = self.external_course_key if use_external_course_key else self.course_id
expected_lms_response = {
'001': 'active',
'002': 'active',
'003': 'inactive'
}
self.mock_course_enrollments_response(self.method, expected_lms_response)

req_data = [
self.student_course_enrollment('active', '001', True),
self.student_course_enrollment('active', '002', False),
self.student_course_enrollment('inactive', '003', True),
]

with self.assert_tracking(
user=self.stem_admin,
program_key=self.cs_program.key,
course_id=course_id,
):
with activate_waffle_flag('enable_course_role_management', self.stem_admin_group):
response = self.request(
self.method, self.get_url(course_id=course_id), self.stem_admin, req_data
)

lms_request_body = json.loads(responses.calls[-1].request.body.decode('utf-8'))
self.assertCountEqual(lms_request_body, [
{
'status': 'active',
'student_key': '001',
'course_staff': True,
},
{
'status': 'active',
'student_key': '002',
'course_staff': False,
},
{
'status': 'inactive',
'student_key': '003',
'course_staff': True,
}
])
self.assertEqual(response.status_code, 200)
self.assertDictEqual(response.data, expected_lms_response)

@mock_oauth_login
@responses.activate
@ddt.data(False, True)
def test_successful_update_course_staff_with_program_group_waffle(self, use_external_course_key):
course_id = self.external_course_key if use_external_course_key else self.course_id
expected_lms_response = {
'001': 'active',
'002': 'active',
'003': 'inactive'
}
self.mock_course_enrollments_response(self.method, expected_lms_response)

req_data = [
self.student_course_enrollment('active', '001', True),
self.student_course_enrollment('active', '002', False),
self.student_course_enrollment('inactive', '003', True),
]

with self.assert_tracking(
user=self.cs_program_admin,
program_key=self.cs_program.key,
course_id=course_id,
):
with activate_waffle_flag('enable_course_role_management', self.cs_program_admin_group):
response = self.request(
self.method, self.get_url(course_id=course_id), self.cs_program_admin, req_data
)

lms_request_body = json.loads(responses.calls[-1].request.body.decode('utf-8'))
self.assertCountEqual(lms_request_body, [
{
'status': 'active',
'student_key': '001',
'course_staff': True,
},
{
'status': 'active',
'student_key': '002',
'course_staff': False,
},
{
'status': 'inactive',
'student_key': '003',
'course_staff': True,
}
])
self.assertEqual(response.status_code, 200)
self.assertDictEqual(response.data, expected_lms_response)

@mock_oauth_login
@responses.activate
@ddt.data(False, True)
@override_flag('enable_course_role_management', active=False)
def test_failed_program_course_enrollment_write_with_course_staff(self, use_external_course_key):
course_id = self.external_course_key if use_external_course_key else self.course_id

req_data = [
self.student_course_enrollment('active', '001', True),
self.student_course_enrollment('active', '002', False),
self.student_course_enrollment('inactive', '003', True),
]

with self.assert_tracking(
user=self.stem_admin,
program_key=self.cs_program.key,
course_id=course_id,
status_code=403,
):
response = self.request(
self.method, self.get_url(course_id=course_id), self.stem_admin, req_data
)

self.assertEqual(response.status_code, 403)

@mock_oauth_login
@responses.activate
@ddt.data(False, True)
Expand Down

0 comments on commit a3783d1

Please sign in to comment.