Commit 98802f72 by Simon Chen

Update people api endpoint permissions

Only those users who belongs to groups with people management permissions will have abilities to create and update instructors
EDUCATOR-1383
parent 4ed6b44a
# pylint: disable=redefined-builtin,no-member
import ddt
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group
from django.contrib.auth.models import Group, Permission
from django.db import IntegrityError
from mock import mock
from rest_framework.reverse import reverse
from rest_framework.test import APITestCase
from testfixtures import LogCapture
from course_discovery.apps.api.tests.mixins import SiteMixin
from course_discovery.apps.api.v1.tests.test_views.mixins import SerializationMixin
from course_discovery.apps.api.v1.tests.test_views.mixins import APITestCase, SerializationMixin
from course_discovery.apps.api.v1.views.people import logger as people_logger
from course_discovery.apps.core.tests.factories import UserFactory
from course_discovery.apps.core.tests.factories import USER_PASSWORD, UserFactory
from course_discovery.apps.course_metadata.models import Person
from course_discovery.apps.course_metadata.people import MarketingSitePeople
from course_discovery.apps.course_metadata.tests import toggle_switch
from course_discovery.apps.course_metadata.tests.factories import OrganizationFactory, PersonFactory, PositionFactory
from course_discovery.apps.publisher.constants import INTERNAL_USER_GROUP_NAME
from course_discovery.apps.publisher.permissions import logger as permission_logger
User = get_user_model()
@ddt.ddt
class PersonViewSetTests(SerializationMixin, SiteMixin, APITestCase):
class PersonViewSetTests(SerializationMixin, APITestCase):
""" Tests for the person resource. """
people_list_url = reverse('api:v1:person-list')
def setUp(self):
super(PersonViewSetTests, self).setUp()
self.user = UserFactory(is_staff=True, is_superuser=True)
self.user.groups.add(Group.objects.get(name=INTERNAL_USER_GROUP_NAME))
self.client.force_authenticate(self.user)
self.user = UserFactory()
self.target_permissions = Permission.objects.filter(
codename__in=['add_person', 'change_person', 'delete_person']
)
internal_test_group = Group.objects.create(name='internal-test')
internal_test_group.permissions.add(*self.target_permissions)
self.user.groups.add(internal_test_group)
self.client.login(username=self.user.username, password=USER_PASSWORD)
self.person = PersonFactory(partner=self.partner)
self.organization = OrganizationFactory(partner=self.partner)
PositionFactory(person=self.person, organization=self.organization)
......@@ -111,26 +110,20 @@ class PersonViewSetTests(SerializationMixin, SiteMixin, APITestCase):
self.client.logout()
Person.objects.all().delete()
response = self.client.post(self.people_list_url, {}, format='json')
self.assertEqual(response.status_code, 403)
self.assertEqual(Person.objects.count(), 0)
response = self.client.post(self.people_list_url)
assert response.status_code == 403
assert Person.objects.count() == 0
def test_create_without_group(self):
def test_create_without_permission(self):
""" Verify group is required when creating a person. """
self.user.groups.remove(Group.objects.get(name=INTERNAL_USER_GROUP_NAME))
self.client.logout()
new_user = UserFactory()
new_user.groups.clear()
self.client.login(username=new_user.username, password=USER_PASSWORD)
current_people_count = Person.objects.count()
with LogCapture(permission_logger.name) as log_capture:
response = self.client.post(self.people_list_url, {}, format='json')
self.assertEqual(response.status_code, 403)
self.assertEqual(Person.objects.count(), current_people_count)
log_capture.check(
(
permission_logger.name,
'INFO',
'Permission denied. User [{}] has no groups'.format(self.user.username),
)
)
response = self.client.post(self.people_list_url)
assert response.status_code == 403
assert Person.objects.count() == current_people_count
def test_get(self):
""" Verify the endpoint returns the details for a single person. """
......@@ -140,6 +133,13 @@ class PersonViewSetTests(SerializationMixin, SiteMixin, APITestCase):
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data, self.serialize_person(self.person))
def test_get_without_authentication(self):
""" Verify the endpoint shows auth error when the details for a single person unauthenticated """
self.client.logout()
url = reverse('api:v1:person-detail', kwargs={'uuid': self.person.uuid})
response = self.client.get(url)
self.assertEqual(response.status_code, 403)
def test_list(self):
""" Verify the endpoint returns a list of all people. """
response = self.client.get(self.people_list_url)
......
......@@ -3,14 +3,13 @@ import logging
import waffle
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework import status, viewsets
from rest_framework.permissions import IsAuthenticated
from rest_framework.permissions import DjangoModelPermissions
from rest_framework.response import Response
from course_discovery.apps.api import filters, serializers
from course_discovery.apps.api.pagination import PageNumberPagination
from course_discovery.apps.course_metadata.exceptions import MarketingSiteAPIClientException, PersonToMarketingException
from course_discovery.apps.course_metadata.people import MarketingSitePeople
from course_discovery.apps.publisher.permissions import UserHasGroup
logger = logging.getLogger(__name__)
......@@ -23,7 +22,7 @@ class PersonViewSet(viewsets.ModelViewSet):
filter_class = filters.PersonFilter
lookup_field = 'uuid'
lookup_value_regex = '[0-9a-f-]+'
permission_classes = (IsAuthenticated, UserHasGroup,)
permission_classes = (DjangoModelPermissions,)
queryset = serializers.PersonSerializer.prefetch_queryset()
serializer_class = serializers.PersonSerializer
pagination_class = PageNumberPagination
......
# -*- coding: utf-8 -*-
# Generated by Django 1.11.3 on 2017-10-20 13:40
from __future__ import unicode_literals
from django.db import migrations
from course_discovery.apps.publisher.constants import (INTERNAL_USER_GROUP_NAME, PARTNER_COORDINATOR_GROUP_NAME,
PUBLISHER_GROUP_NAME, REVIEWER_GROUP_NAME)
GROUPS = [INTERNAL_USER_GROUP_NAME, PARTNER_COORDINATOR_GROUP_NAME, REVIEWER_GROUP_NAME, PUBLISHER_GROUP_NAME]
def add_people_permission(apps, schema_editor):
Permission = apps.get_model('auth', 'Permission')
OrganizationExtension = apps.get_model('publisher', 'OrganizationExtension')
Group = apps.get_model('auth', 'Group')
target_permissions = Permission.objects.filter(
codename__in=['add_person', 'change_person', 'delete_person']
)
for org in OrganizationExtension.objects.all():
org.group.permissions.add(*target_permissions)
# Also add the permissions to publisher role groups
publisher_groups = Group.objects.filter(name__in=GROUPS)
for group in publisher_groups:
group.permissions.add(*target_permissions)
def remove_people_permission(apps, schema_editor):
Permission = apps.get_model('auth', 'Permission')
OrganizationExtension = apps.get_model('publisher', 'OrganizationExtension')
Group = apps.get_model('auth', 'Group')
target_permissions = Permission.objects.filter(
codename__in=['add_person', 'change_person', 'delete_person']
)
for org in OrganizationExtension.objects.all():
org.group.permissions.remove(*target_permissions)
# Also remove the permissions to publisher role groups
publisher_groups = Group.objects.filter(name__in=GROUPS)
for group in publisher_groups:
group.permissions.remove(*target_permissions)
class Migration(migrations.Migration):
dependencies = [
('publisher', '0060_auto_20171004_0521'),
]
operations = [
migrations.RunPython(add_people_permission, remove_people_permission),
]
import logging
from rest_framework import permissions
logger = logging.getLogger(__name__)
class UserHasGroup(permissions.BasePermission):
"""
Global permission to check if request.user has any group
"""
def has_permission(self, request, view):
if request.user.groups.all():
return True
logger.info('Permission denied. User [%s] has no groups', request.user.username)
return False
import logging
import waffle
from django.contrib.auth.models import Permission
from django.core.exceptions import ObjectDoesNotExist
from django.db.models.signals import post_save
from django.dispatch import receiver
from slumber.exceptions import SlumberBaseException
from course_discovery.apps.publisher.models import CourseRun
from course_discovery.apps.publisher.models import CourseRun, OrganizationExtension
from course_discovery.apps.publisher.studio_api_utils import StudioAPI
logger = logging.getLogger(__name__)
......@@ -72,3 +73,12 @@ def create_course_run_in_studio_receiver(sender, instance, created, **kwargs):
logger.exception('Failed to update Studio image for course run [%s]', instance.lms_course_id)
logger.info('Completed creation of course run [%s] on Studio.', instance.lms_course_id)
@receiver(post_save, sender=OrganizationExtension)
def add_permissions_to_organization_group(sender, instance, created, **kwargs): # pylint: disable=unused-argument
if created:
target_permissions = Permission.objects.filter(
codename__in=['add_person', 'change_person', 'delete_person']
)
instance.group.permissions.add(*target_permissions)
......@@ -4,6 +4,7 @@ import json
import mock
import pytest
import responses
from django.contrib.auth.models import Permission
from freezegun import freeze_time
from slumber.exceptions import HttpServerError
from waffle.testutils import override_switch
......@@ -211,3 +212,15 @@ class TestCreateCourseRunInStudio:
publisher_course_run.id,
json.dumps(body).encode('utf8')
)
@pytest.mark.django_db
class TestCreateOrganizations:
def test_create_organizations_added_permissions(self):
# Make sure created organization automatically have people permissions
organization = OrganizationExtensionFactory()
target_permissions = Permission.objects.filter(
codename__in=['add_person', 'change_person', 'delete_person']
)
for permission in target_permissions:
assert organization.group.permissions.filter(codename=permission.codename).exists()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment