Commit 2c3421ff authored by Daniel Schreiber's avatar Daniel Schreiber

add api for csv uploads

parent 08fd406b
Pipeline #28823 failed with stage
in 46 seconds
# -*- coding: utf-8 -*-
from __future__ import unicode_literals, print_function
import json
from datetime import datetime
from datetime import timedelta
import uuid
from unittest import skip
from django.core.files.uploadedfile import SimpleUploadedFile
from django.conf import settings
from django.test import override_settings
from django.utils import timezone
from django.test.client import encode_multipart
from rest_framework.renderers import JSONRenderer
from rest_framework.reverse import reverse
from rest_framework.test import APITestCase
......@@ -296,3 +300,97 @@ class ServiceProviderGetTest(APITestCase):
url = reverse("service_detail", args=(self.local_sp.pk,))
r = self.client.get(url)
self.assertEqual(r.status_code, 200)
class CsvUploadViewTest(APITestCase):
fixtures = ['csv_import_data.json']
def upload_csv(self, data):
csv_data = "\n".join([
"{},{}".format(d[0].eppn, d[1]) for d in data
]).encode("utf-8")
upload = SimpleUploadedFile('data.csv', csv_data, 'text/csv')
data = {'csv': upload}
content = encode_multipart('BoUnDaRyStRiNg', data)
content_type = 'multipart/form-data; boundary=BoUnDaRyStRiNg'
#print(content)
r = self.client.post(self.url, content, content_type=content_type)
return r
def setUp(self):
self.user1 = Resource.objects.get(eppn="user1@a.edu")
self.user2 = Resource.objects.get(eppn="user2@a.edu")
self.user3 = Resource.objects.get(eppn="user3@a.edu")
self.user_extern = Resource.objects.get(eppn="user3@b.edu")
self.idp_user = TokenAuthUser.objects.create_superuser(
username='api',
email='foo@bar.com',
password='api',
user_type=UserType.TYPE_API,
api_url="http://api.a.edu:7000/",
realm="a.edu",
)
self.url = reverse('csv_upload')
def test_upload_unauthenticated(self):
now = timezone.now()
data = (
(self.user1, '31.12.%s' % now.year),
)
r = self.upload_csv(data)
self.assertEqual(r.status_code, 401)
def test_modify(self):
now = timezone.now()
data = (
(self.user1, '31.12.%s' % now.year),
)
self.client.force_authenticate(self.idp_user)
r = self.upload_csv(data)
self.assertEqual(r.status_code, 200)
self.assertEqual(
Resource.objects.get(pk=self.user1.pk).expiry_date,
timezone.make_aware(datetime.strptime("%s-12-31T00:00:00" % now.year, "%Y-%m-%dT%H:%M:%S"))
)
def test_do_not_modify(self):
self.assertEqual(0, QueueJob.objects.count())
now = timezone.now()
today = timezone.make_aware(datetime.strptime("{}-{}-{}T00:00:00".format(now.year, now.month, now.day), "%Y-%m-%dT%H:%M:%S"))
Resource.objects.update(expiry_date=today, deletion_date=today + timedelta(days=30))
self.user1 = Resource.objects.get(eppn="user1@a.edu")
u1_ex = self.user1.expiry_date
data = [
(res, res.expiry_date.astimezone(timezone.get_default_timezone()).strftime("%d.%m.%Y"))
for res in Resource.objects.filter(realm_idp=settings.SERVER_REALM)
]
self.client.force_authenticate(self.idp_user)
r = self.upload_csv(data)
self.assertEqual(r.status_code, 200)
self.assertEqual(u1_ex, Resource.objects.get(pk=self.user1.pk).expiry_date)
self.assertEqual(0, QueueJob.objects.count())
def test_dont_touch_external_users(self):
data = (
(self.user_extern, '31.12.2016'),
)
self.client.force_authenticate(self.idp_user)
r = self.upload_csv(data)
self.assertEqual(r.status_code, 200)
self.assertEqual(
Resource.objects.get(pk=self.user_extern.pk).expiry_date,
self.user_extern.expiry_date
)
def test_trigger_sync_for_external_services(self):
data = (
(self.user3, '31.12.2016'),
)
self.client.force_authenticate(self.idp_user)
r = self.upload_csv(data)
self.assertEqual(r.status_code, 200)
self.assertEqual(1, QueueJob.objects.count())
q = QueueJob.objects.first()
self.assertEqual(q.res_uuid, self.user3.pk)
......@@ -9,6 +9,7 @@ from api import views
urlpatterns = [
url(r'^$', views.WelcomeView.as_view()),
url(r'^health/$', views.Health.as_view(), name='health'),
url(r'^csv_upload/$', views.CsvUploadView.as_view(), name='csv_upload'),
url(r'^services/$', views.ServiceList.as_view(), name='service_list'),
url(r'^services/(?P<uuid>[^/]+)/$',
views.ServiceDetail.as_view(), name='service_detail'),
......
......@@ -23,7 +23,7 @@ from rest_framework import status
from rest_framework.authentication import TokenAuthentication
from rest_framework.authtoken.models import Token
from rest_framework.exceptions import PermissionDenied, ParseError
from rest_framework.parsers import JSONParser
from rest_framework.parsers import JSONParser, MultiPartParser
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
......@@ -31,6 +31,7 @@ from rest_framework.views import APIView
from api.models import Resource, ServiceProvider, Attribute, UserType
from api.serializers import ServiceSerializer, ResourceSerializer, AttributeSerializer
from api.signals import resource_created
from api.utils.csv_handling import sync_from_csv_data
from api.utils.queue_utils import dispatch_resource_sync, dispatch_deletion_sync
from api.utils.user_utils import local_reverse
from api.utils.validation_utils import validate_uuid4
......@@ -261,3 +262,17 @@ class ResourceAttributes(AuthAPIView):
class WelcomeView(APIView):
def get(self, request):
return Response(data="FRMS-API-0.1")
class CsvUploadView(AuthAPIView):
parser_classes = [MultiPartParser]
def post(self, request):
if request.user.user_type != UserType.TYPE_API:
raise PermissionDenied()
csv_file = request.data.get('csv', None)
if not csv_file:
return Response("csv parameter required", status=Response.HTTP_400_BAD_REQUEST)
date_format = request.data.get('date_format', '%d.%m.%Y')
sync_from_csv_data(csv_file, date_format)
return Response('ok')
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