diff --git a/registrar/apps/api/v1/mixins.py b/registrar/apps/api/v1/mixins.py index eeed8f53..28af781c 100644 --- a/registrar/apps/api/v1/mixins.py +++ b/registrar/apps/api/v1/mixins.py @@ -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 @@ -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, @@ -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') @@ -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') diff --git a/registrar/apps/api/v1/tests/test_views.py b/registrar/apps/api/v1/tests/test_views.py index 75b1f2f4..b191f7ce 100644 --- a/registrar/apps/api/v1/tests/test_views.py +++ b/registrar/apps/api/v1/tests/test_views.py @@ -3,6 +3,7 @@ import json import logging import uuid +from contextlib import contextmanager from io import StringIO from posixpath import join as urljoin @@ -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 @@ -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 """ @@ -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() @@ -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): @@ -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)