API modificacion de usuario

musiclist/settings/__init__.py

@@ -44,7 +44,6 @@ MIDDLEWARE = [
     'django.contrib.auth.middleware.AuthenticationMiddleware',
     'django.contrib.messages.middleware.MessageMiddleware',
     'django.middleware.clickjacking.XFrameOptionsMiddleware',
-    'utils.middleware.PUTParsingMiddleware',
 ]
 
 """Root import path to urlconf"""

test.sh

@@ -1,3 +1,3 @@
 #!/bin/sh
 
-coverage run --source='.' --omit='venv/*' manage.py test --failfast
+coverage run --source='.' --omit '.venv/*' manage.py test

users/api_urls.py

@@ -3,9 +3,6 @@ from django.urls import path
 from users import api_views
 
 urlpatterns = [
-    path('user/', api_views.user),
-    path('user/<int:user_id>', api_views.user),
-
-    path('user/social_networks', api_views.social_networks),
-    path('user/<int:user_id>/social_networks', api_views.social_networks)
+    path('user/<int:user_id>', api_views.user_view),
+    path('user/<int:user_id>/social_networks', api_views.social_networks_view)
 ]

users/api_views.py

@@ -1,27 +1,28 @@
+import json
+
 from django.http import JsonResponse
-from django.views.decorators.http import require_GET
 from oauth2_provider.decorators import protected_resource
 
-from users.forms import SocialNetworksForm
+from users.forms import SocialNetworksForm, UserForm
 from users.models import User, SocialNetworks
 
 
-@protected_resource()
-def user(request, user_id=None):
+def user_view(request, user_id=None):
+    if user_id is None:
+        return JsonResponse({'status': 400, 'error': 'No se entrego un user_id'}, status=400)
+
+    user = User.objects.filter(pk=user_id)
+    if user.count() != 1:
+        return JsonResponse({'status': 404, 'error': f'No existe un usuario con id {user_id}'}, status=404)
+    user = user[0]
+
     if request.method == 'GET':
-        return _get_user(request, user_id)
+        return _get_user(request, user)
+    if request.method == 'PUT':
+        return _put_user(request, user)
 
 
-@require_GET
-def _get_user(request, user_id=None):
-    user = request.user
-
-    if user_id is not None:
-        user = User.objects.filter(pk=user_id)
-        if user.count() != 1:
-            return JsonResponse({'status': 404, 'error': f'No existe un usuario con id {user_id}'})
-        user = user[0]
-
+def _get_user(request, user):
     encoded_user = {
         'id': user.id,
         'username': user.username,
@@ -32,23 +33,41 @@ def _get_user(request, user_id=None):
 
 
 @protected_resource()
-def social_networks(request, user_id=None):
+def _put_user(request, user):
+    if request.user.id != user.id and not request.user.is_admin:
+        return JsonResponse({'status': 403,
+                             'error': 'El usuario no tiene permiso para hacer esta acción'},
+                            status=403)
+
+    request_data = json.loads(request.body.decode('utf8'))
+    form = UserForm(request_data, instance=user)
+
+    if not form.is_valid():
+        return JsonResponse({'status': 400, 'error': form.errors.as_json()}, status=400)
+
+    form.save()
+
+    return JsonResponse({'status': 200}, status=200)
+
+
+def social_networks_view(request, user_id=None):
     if user_id is None:
-        user_id = request.user.id
+        return JsonResponse({'status': 400, 'error': 'No se entrego un user_id'}, status=400)
 
-    if request.method == 'GET':
-        return _get_social_networks(request, user_id)
-    if request.method == 'PUT':
-        return _update_social_networks(request, user_id)
-
-
-def _get_social_networks(request, user_id):
     social_networks = SocialNetworks.objects.filter(user_id=user_id)
     if social_networks.count() != 1:
-        return JsonResponse({'status': 404, 'error': f'No existe redes sociales de un usuario con id {user_id}'})
+        return JsonResponse({'status': 404, 'error': f'No existe redes sociales de un usuario con id {user_id}'},
+                            status=404)
 
     social_networks = social_networks[0]
 
+    if request.method == 'GET':
+        return _get_social_networks(request, social_networks)
+    if request.method == 'PUT':
+        return _update_social_networks(request, social_networks)
+
+
+def _get_social_networks(request, social_networks):
     encoded_social_networks = {
         'twitter': social_networks.twitter,
         'facebook': social_networks.facebook,
@@ -56,26 +75,22 @@ def _get_social_networks(request, user_id):
         'youtube': social_networks.youtube,
         'twitch': social_networks.twitch
     }
+
     return JsonResponse(encoded_social_networks)
 
 
-def _update_social_networks(request, user_id=None):
-    social_networks = SocialNetworks.objects.filter(user_id=user_id)
-    if social_networks.count() != 1:
-        return JsonResponse({'status': 404, 'error': f'No existe redes sociales de un usuario '
-                                                     f'con id {user_id}'}, status=404)
+@protected_resource()
+def _update_social_networks(request, social_networks):
+    if request.user.id != social_networks.user_id and not request.user.is_admin:
+        return JsonResponse({'status': 403,
+                             'error': 'El usuario no tiene permiso para hacer esta acción'},
+                            status=403)
 
-    social_networks = social_networks[0]
+    request_data = json.loads(request.body.decode('utf8'))
+    form = SocialNetworksForm(request_data, instance=social_networks)
 
-    form = SocialNetworksForm(request.PUT or None)
     if not form.is_valid():
-        return JsonResponse({'status': 400, 'errors': form.errors}, status=400)
-
-    social_networks.twitter = form.cleaned_data['twitter']
-    social_networks.facebook = form.cleaned_data['facebook']
-    social_networks.instagram = form.cleaned_data['instagram']
-    social_networks.youtube = form.cleaned_data['youtube']
-    social_networks.twitch = form.cleaned_data['twitch']
-    social_networks.save()
+        return JsonResponse({'status': 400, 'error': form.errors.as_json()}, status=400)
 
+    form.save()
     return JsonResponse({'status': 200}, status=200)

users/forms.py

@@ -1,9 +1,14 @@
 from django import forms
+from users.models import User, SocialNetworks
 
 
-class SocialNetworksForm(forms.Form):
-    twitter = forms.CharField(max_length=255, strip=True, required=False, empty_value='')
-    facebook = forms.CharField(max_length=255, strip=True, required=False, empty_value='')
-    instagram = forms.CharField(max_length=255, strip=True, required=False, empty_value='')
-    youtube = forms.CharField(max_length=255, strip=True, required=False, empty_value='')
-    twitch = forms.CharField(max_length=255, strip=True, required=False, empty_value='')
+class UserForm(forms.ModelForm):
+    class Meta:
+        model = User
+        fields = ('username', 'email')
+
+
+class SocialNetworksForm(forms.ModelForm):
+    class Meta:
+        model = SocialNetworks
+        fields = ('twitter', 'facebook', 'instagram', 'youtube', 'twitch')

users/test.py

@@ -0,0 +1,136 @@
+import json
+
+from django.test import TestCase
+
+from users.models import User, SocialNetworks
+from django.utils import timezone
+from datetime import timedelta
+
+
+class TestAPIViews(TestCase):
+    def setUp(self):
+        self.user_admin = User.objects.create_superuser('admin', 'email@email.com', 'superpassword')
+        self.user_regular = User.objects.create_user('normal', 'email@email.com', 'superpassword')
+
+        tomorrow = timezone.now() + timedelta(1)
+
+        self.user_token = self.user_regular.oauth2_provider_accesstoken.create(expires=tomorrow, token='usertoken')
+        self.admin_token = self.user_admin.oauth2_provider_accesstoken.create(expires=tomorrow, token='admintoken')
+
+        self.user_regular.socialnetworks.twitter = 'twitter'
+        self.user_regular.socialnetworks.save()
+
+    def _admin_header(self):
+        return f'Bearer {self.admin_token.token}'
+
+    def _user_header(self):
+        return f'Bearer {self.user_token.token}'
+
+    def test_get_user_exists(self):
+        response = self.client.get('/api/users/user/1', follow=True)
+        self.assertEqual(response.status_code, 200)
+
+    def test_get_user_not_exists(self):
+        response = self.client.get('/api/users/user/3', follow=True)
+        self.assertEqual(response.status_code, 404)
+
+    def test_put_user_valid(self):
+        new_user_data = {
+            'username': 'other',
+            'email': 'skrd159@gmail.com'
+        }
+
+        response = self.client.put('/api/users/user/2', json.dumps(new_user_data),
+                                   HTTP_AUTHORIZATION=self._user_header())
+
+        self.assertEqual(response.status_code, 200)
+
+        user = User.objects.get(pk=2)
+        self.assertEqual(user.username, new_user_data['username'])
+        self.assertEqual(user.email, new_user_data['email'])
+
+    def test_put_user_not_all_data(self):
+        new_user_data = {
+            'username': 'other',
+        }
+
+        response = self.client.put('/api/users/user/2', json.dumps(new_user_data),
+                                   HTTP_AUTHORIZATION=self._user_header())
+
+        self.assertEqual(response.status_code, 400)
+
+    def test_put_user_not_modifying_self(self):
+        new_user_data = {
+            'username': 'other',
+            'email': 'skrd159@gmail.com'
+        }
+
+        response = self.client.put('/api/users/user/1', json.dumps(new_user_data),
+                                   HTTP_AUTHORIZATION=self._user_header())
+
+        self.assertEqual(response.status_code, 403)
+
+    def test_put_user_admin_not_modifying_self(self):
+        new_user_data = {
+            'username': 'other',
+            'email': 'skrd159@gmail.com'
+        }
+
+        response = self.client.put('/api/users/user/2', json.dumps(new_user_data),
+                                   HTTP_AUTHORIZATION=self._admin_header())
+
+        self.assertEqual(response.status_code, 200)
+
+    def test_get_social_networks_exists(self):
+        response = self.client.get('/api/users/user/2/social_networks', follow=True)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json()['twitter'], 'twitter')
+
+    def test_get_social_networks_not_exists(self):
+        response = self.client.get('/api/users/user/3/social_networks', follow=True)
+        self.assertEqual(response.status_code, 404)
+
+    def test_put_social_networks_valid(self):
+        new_social_data = {
+            'twitter': 'ryuuji159',
+            'facebook': '',
+            'instagram': '',
+            'youtube': '',
+            'twitch': '',
+        }
+
+        response = self.client.put('/api/users/user/2/social_networks', json.dumps(new_social_data),
+                                   HTTP_AUTHORIZATION=self._user_header())
+
+        self.assertEqual(response.status_code, 200)
+
+        social_networks = SocialNetworks.objects.get(user_id=2)
+        self.assertEqual(social_networks.twitter, new_social_data['twitter'])
+        self.assertEqual(social_networks.facebook, new_social_data['facebook'])
+        self.assertEqual(social_networks.instagram, new_social_data['instagram'])
+        self.assertEqual(social_networks.youtube, new_social_data['youtube'])
+        self.assertEqual(social_networks.twitch, new_social_data['twitch'])
+
+    def test_put_social_networks_empty(self):
+        new_social_data = {}
+
+        response = self.client.put('/api/users/user/2/social_networks', json.dumps(new_social_data),
+                                   HTTP_AUTHORIZATION=self._user_header())
+
+        self.assertEqual(response.status_code, 200)
+
+    def test_put_social_networks_not_modifying_self(self):
+        new_social_data = {}
+
+        response = self.client.put('/api/users/user/1/social_networks', json.dumps(new_social_data),
+                                   HTTP_AUTHORIZATION=self._user_header())
+
+        self.assertEqual(response.status_code, 403)
+
+    def test_put_social_networks_admin_modifying_other(self):
+        new_social_data = {}
+
+        response = self.client.put('/api/users/user/2/social_networks', json.dumps(new_social_data),
+                                   HTTP_AUTHORIZATION=self._admin_header())
+
+        self.assertEqual(response.status_code, 200)

utils/middleware.py

@@ -1,33 +0,0 @@
-class PUTParsingMiddleware:
-    """ Fuerza a django a parsear el body de una request con método PUT
-
-    Django no parsea este tipo de request, dejando todo en el body, esto es por varias razones
-    aparentemente validas explicadas aquí
-    "https://groups.google.com/forum/#!msg/django-developers/dxI4qVzrBY4/m_9IiNk_p7UJ"
-    asi que no hay nada que hacerle, sin embargo, para el caso de uso de esta
-    aplicación, se puede asumir que sera seguro y funcionara correctamente.
-
-    Este código lo saque el blog post "https://thihara.github.io/Django-Req-Parsing/"
-    """
-
-    def __init__(self, func):
-        self.func = func
-
-    def __call__(self, request, *args, **kwargs):
-        if request.method == 'PUT' and request.content_type == 'application/x-www-form-urlencoded':
-            if hasattr(request, '_post'):
-                del request._post
-                del request._files
-
-            try:
-                request.method = "POST"
-                request._load_post_and_files()
-                request.method = "PUT"
-            except AttributeError:
-                request.META['REQUEST_METHOD'] = 'POST'
-                request._load_post_and_files()
-                request.META['REQUEST_METHOD'] = 'PUT'
-
-            request.PUT = request.POST
-
-        return self.func(request, *args, **kwargs)