added custom user endpoints

This commit is contained in:
Sam
2021-01-25 12:15:11 +00:00
parent b2c857878b
commit f94d822d5e
6 changed files with 259 additions and 1 deletions

View File

@@ -1,5 +1,6 @@
from rest_framework import routers from rest_framework import routers
from core.views import CustomUserViewSet
from companies.views import CompanyViewSet from companies.views import CompanyViewSet
from products.views import ProductViewSet from products.views import ProductViewSet
from history.views import HistorySyncViewSet from history.views import HistorySyncViewSet
@@ -9,6 +10,8 @@ from geo import views as geo_views
router = routers.DefaultRouter() router = routers.DefaultRouter()
router.register('users', CustomUserViewSet, basename='users')
router.register('companies', CompanyViewSet, basename='company') router.register('companies', CompanyViewSet, basename='company')
router.register('products', ProductViewSet, basename='product') router.register('products', ProductViewSet, basename='product')
router.register('history', HistorySyncViewSet, basename='history') router.register('history', HistorySyncViewSet, basename='history')

View File

@@ -26,6 +26,7 @@ class UserManager(BaseUserManager):
def create_superuser(self, email, password, **extra_fields): def create_superuser(self, email, password, **extra_fields):
extra_fields.setdefault('is_superuser', True) extra_fields.setdefault('is_superuser', True)
extra_fields.setdefault('is_staff', True) extra_fields.setdefault('is_staff', True)
extra_fields.setdefault('is_active', True)
if extra_fields.get('is_superuser') is not True: if extra_fields.get('is_superuser') is not True:
raise ValueError('Superuser must have is_superuser=True.') raise ValueError('Superuser must have is_superuser=True.')

17
core/serializers.py Normal file
View File

@@ -0,0 +1,17 @@
from rest_framework import serializers
from . import models
class CustomUserSerializer(serializers.ModelSerializer):
class Meta:
model = models.CustomUser
fields = ('email', 'full_name', 'role', 'is_active')
class CreatorSerializer(serializers.ModelSerializer):
class Meta:
model = models.CustomUser
fields = ('email',)

View File

@@ -1,3 +1,224 @@
import random
import string
from django.test import TestCase from django.test import TestCase
from rest_framework.test import APITestCase
from rest_framework import status
from core.utils import get_tokens_for_user
from . import models
from . import factories
# Create your tests here. # Create your tests here.
class CustomUserViewSetTest(APITestCase):
"""CustomUser viewset tests
"""
def setUp(self):
"""Tests setup
"""
self.endpoint = '/api/v1/users/'
self.factory = factories.CustomUserFactory
self.model = models.CustomUser
# create admin user
self.admin_email = f"admin_user@mail.com"
self.password = ''.join(random.choices(string.ascii_uppercase, k = 10))
self.user = self.factory(email=self.admin_email, password=self.password, is_active=True)
# create regular user
self.reg_email = f"regular_user@mail.com"
self.password = ''.join(random.choices(string.ascii_uppercase, k = 10))
self.user = self.factory(email=self.reg_email, password=self.password, is_active=True)
# anon user
def test_anon_user_can_create_inactive_instance(self):
"""Not logged-in user can create new instance of User but it's inactive
TODO: should create inactive user
"""
# Query endpoint
response = self.client.post(self.endpoint, data={})
# Assert access is forbidden
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
def test_anon_user_cannot_modify_existing_instance(self):
"""Not logged-in user cannot modify existing instance
"""
# Create instance
instance = self.factory()
# Query endpoint
url = self.endpoint + f'{instance.pk}/'
response = self.client.put(url, {}, format='json')
# Assert forbidden code
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
def test_anon_user_cannot_delete_existing_instance(self):
"""Not logged-in user cannot delete existing instance
"""
# Create instances
instance = self.factory()
# Query endpoint
url = self.endpoint + f'{instance.pk}/'
response = self.client.delete(url)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
# Assert instance still exists on db
self.assertTrue(self.model.objects.get(id=instance.pk))
def test_anon_user_cannot_list_instances(self):
"""Not logged-in user can't read instance
"""
# Request list
response = self.client.get(self.endpoint)
# Assert access is forbidden
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
# auth regular user
def test_regular_user_cannot_create_instance(self):
"""Regular user cannot create new instance
"""
# Authenticate
token = get_tokens_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
# Query endpoint
response = self.client.post(self.endpoint, data={})
# Assert access is forbidden
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_regular_user_cannot_modify_existing_instance(self):
"""Regular user cannot modify existing instance
"""
# Create instance
instance = self.factory()
# Authenticate
token = get_tokens_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
# Query endpoint
url = self.endpoint + f'{instance.pk}/'
response = self.client.put(url, {}, format='json')
# Assert forbidden code
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_regular_user_cannot_delete_existing_instance(self):
"""Regular user cannot delete existing instance
"""
# Create instances
instance = self.factory()
# Authenticate
token = get_tokens_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
# Query endpoint
url = self.endpoint + f'{instance.pk}/'
response = self.client.delete(url)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
# Assert instance still exists on db
self.assertTrue(self.model.objects.get(id=instance.pk))
def test_regular_user_cannot_list_instances(self):
"""Regular user can't list instances
"""
# Request list
response = self.client.get(self.endpoint)
# Authenticate
token = get_tokens_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
# Assert access is forbidden
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
# admin user
def test_admin_user_can_create_instance(self):
"""Admin user can create new instance
"""
# set user as admin
self.user.is_staff = True
self.user.save()
# Define request data
data = {
'email': 'test@email.com',
'full_name': 'TEST NAME',
'password1': 'VENTILADORES1234499.89',
'password2': 'VENTILADORES1234499.89',
}
# Authenticate user
token = get_tokens_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
# Query endpoint
response = self.client.post(self.endpoint, data=data, format='json')
# Assert endpoint returns created status
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
# Assert instance exists on db
self.assertTrue(self.model.objects.get(email=response.data['email']))
def test_admin_user_can_modify_existing_instance(self):
"""Admin user can modify existing instance
"""
# set user as admin
self.user.is_staff = True
self.user.save()
# Create instances
instance = self.factory()
# Define request data
data = {
'email': 'test_alt@email.com',
'full_name': 'TEST NAME ALT'
}
# Authenticate user
token = get_tokens_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
# Query endpoint
url = self.endpoint + f'{instance.pk}/'
response = self.client.put(url, data, format='json')
# Assert endpoint returns OK code
self.assertEqual(response.status_code, status.HTTP_200_OK)
# Assert instance has been modified
for key in data:
self.assertEqual(data[key], response.data[key])
def test_admin_user_can_delete_existing_instance(self):
"""Admin user can delete existing instance
"""
# set user as admin
self.user.is_staff = True
self.user.save()
# Create instances
instance = self.factory()
# Authenticate user
token = get_tokens_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
# Query endpoint
url = self.endpoint + f'{instance.pk}/'
response = self.client.delete(url)
# assert 204 no content
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
# Assert instance doesn't exists anymore on db
self.assertFalse(self.model.objects.filter(id=instance.pk).exists())

View File

@@ -1,3 +1,19 @@
from django.shortcuts import render from django.shortcuts import render
from django.http import HttpResponse
from rest_framework import status
from rest_framework import viewsets
from rest_framework.response import Response
from rest_framework.permissions import IsAdminUser
from . import models
from . import serializers
# Create your views here. # Create your views here.
class CustomUserViewSet(viewsets.ModelViewSet):
model = models.CustomUser
serializer_class = serializers.CustomUserSerializer
model_name = 'custom_user'
queryset = models.CustomUser.objects.all()
permission_classes = [IsAdminUser,]

View File

@@ -30,7 +30,7 @@ class ProductViewSetTest(APITestCase):
self.password = ''.join(random.choices(string.ascii_uppercase, k = 10)) self.password = ''.join(random.choices(string.ascii_uppercase, k = 10))
self.user = CustomUserFactory(password=self.password) self.user = CustomUserFactory(password=self.password)
# user not authenticated # anon user
def test_anon_user_cannot_create_instance(self): def test_anon_user_cannot_create_instance(self):
"""Not logged-in user cannot create new instance """Not logged-in user cannot create new instance
""" """