more work on load_coop_managers endpoint, unfinished

back_latienda/urls.py

@@ -31,5 +31,6 @@ urlpatterns = [
     path('api/v1/token/verify/', TokenVerifyView.as_view(), name='token_verify'),
     path('api/v1/user/change_password/<int:pk>/', core_views.ChangeUserPasswordView.as_view(), name="change-password"),
     path('api/v1/user/update/<int:pk>/', core_views.UpdateUserView.as_view(), name="update-user"),
+    path('api/v1/load_coops/', core_views.load_coop_managers, name='csv-loader'),
     path('api/v1/', include(router.urls)),
 ] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)

core/tests.py

@@ -3,14 +3,19 @@ import string
 import json
 import hashlib
 import base64
+import csv
 
 from django.test import TestCase
 
 from rest_framework.test import APITestCase
 from rest_framework import status
 
+import requests
+
 from core.utils import get_tokens_for_user
 
+from companies.models import Company
+
 from . import models
 from . import factories
 # Create your tests here.
@@ -376,3 +381,82 @@ class UpdateUserViewTest(APITestCase):
         # Assert forbidden code
         self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
 
+
+class LoadCoopManagerTestCase(TestCase):
+
+    def setUp(self):
+        """Tests setup
+        """
+        self.endpoint = 'http://127.0.0.1:8000/api/v1/load_coops/'
+        self.user_factory = factories.CustomUserFactory
+        self.user_model = models.CustomUser
+        self.company_model = Company
+        # create admin user
+        self.admin_email = f"admin_user@mail.com"
+        self.password = ''.join(random.choices(string.ascii_uppercase, k = 10))
+        self.admin_user = self.user_factory(email=self.admin_email, password=self.password, is_staff=True, is_active=True)
+        # create regular user
+        self.reg_email = f"user@mail.com"
+        self.user = self.user_factory(email=self.reg_email, is_active=True)
+        self.user.set_password(self.password)
+        self.user.save()
+
+    def test_admin_can_load_csv(self):
+        company_count = self.company_model.objects.count()
+        user_count = self.user_model.objects.count()
+
+        # read csv file
+        csv_file = '../coop.csv'
+        files = {'csv_file': open(csv_file,'rt')}
+
+        # Authenticate
+        token = get_tokens_for_user(self.admin_user)
+        headers = {'Authorization': f"Bearer {token['access']}"}
+
+        # send in request
+        response = requests.post(self.endpoint, files=files, headers=headers)
+
+        # check response
+
+        # check for object creation
+        self.assertNotEqual(company_count, self.company_model.objects.count())
+        self.assertNotEqual(user_count, self.user_model.objects.count())
+
+    def test_auth_user_cannot_load_csv(self):
+        company_count = self.company_model.objects.count()
+        user_count = self.user_model.objects.count()
+
+        # read csv file
+        csv_file = '../coop.csv'
+        files = {'csv_file': open(csv_file,'r')}
+
+        # Authenticate
+        token = get_tokens_for_user(self.user)
+        headers = {'Authorization': f"Bearer {token['access']}"}
+
+        # send in request
+        response = requests.post(self.endpoint, files=files, headers=headers)
+
+        # check response
+
+        # check for object creation
+        self.assertEqual(company_count, self.company_model.objects.count())
+        self.assertEqual(user_count, self.user_model.objects.count())
+
+    def test_anon_user_cannot_load_csv(self):
+        company_count = self.company_model.objects.count()
+        user_count = self.user_model.objects.count()
+
+        # read csv file
+        csv_file = '../coop.csv'
+        files = {'csv_file': open(csv_file,'r')}
+
+        # Query endpoint
+        response = self.client.post(self.endpoint, files=files)
+
+        # check response
+
+        # check for object creation
+        self.assertEqual(company_count, self.company_model.objects.count())
+        self.assertEqual(user_count, self.user_model.objects.count())
+

core/views.py

@@ -1,5 +1,7 @@
 import csv
 import logging
+import json
+import io
 
 from django.shortcuts import render
 from django.http import HttpResponse
@@ -12,6 +14,8 @@ from rest_framework.permissions import IsAdminUser
 from rest_framework.generics import UpdateAPIView
 from rest_framework.decorators import api_view, permission_classes
 
+from companies.models import Company
+
 from . import models
 from . import serializers
 
@@ -87,23 +91,40 @@ def load_coop_managers(request):
     """Read CSV file being received
     Parse it to create users and related companies
     """
-    csv_file = request.FILES["csv_file"]
+    try:
-    if csv_file.name.endswith('.csv') is not True:
-        logging.error(f"File {csv_file.name} is not a CSV file")
-        return Response({"errors":{"details": "File is not CSV type"}})
 
-    logging.info(f"Reading contents of  {csv_file.name}")
+        csv_file = request.FILES['csv_file']
-    csv_reader =  csv.DictReader(csv_file, delimiter=',')
+        if csv_file.name.endswith('.csv') is not True:
-    for row in csv_reader:
+            logging.error(f"File {csv_file.name} is not a CSV file")
-        email = row['email']
+            return Response({"errors":{"details": "File is not CSV type"}})
-        cif = row['cif']
-        company_name = row['nombre-coop']
-        short_name = row['nombre-corto']
-        url = row['url']
-        shop = row['es-tienda']
 
-        coop = None
+        logging.info(f"Reading contents of  {csv_file.name}")
-        coop_user = User.objects.create_user(email=email, company=coop, role='COOP_MANAGER')
+        contents = csv_file.file
+        import ipdb; ipdb.set_trace()
+        csv_reader =  csv.reader(contents, delimiter=',')
+        coop_counter = 0
+        user_counter = 0
+        for row in csv_reader:
+            import ipdb; ipdb.set_trace()
+            try:
+                coop_data = {
+                    'cif': row['cif'],
+                    'company_name': row['nombre-coop'],
+                    'short_name': row['nombre-corto'],
+                    'shop': row['es-tienda'],
+                    'url': row['url'],
+                }
 
+                coop = Company.object.create(**coop_data)
+                logging.info(f"Created Coop: {coop_data}")
+                coop_counter += 1
 
-    return Response
+                coop_user = User.objects.create_user(email=row['email'], company=coop, role='COOP_MANAGER')
+                logging.info(f"Created User: {coop_user}")
+                user_counter += 1
+            except Exception as e:
+                logging.error(f"Could not parse {row}")
+
+        return Response()
+    except Exception as e:
+        return Response({"errors": {"details": str(type(e))}})

requirements.txt

@@ -10,3 +10,5 @@ django-cors-headers==3.5.0
 django-taggit-serializer==0.1.7
 django-tagulous==1.1.0
 Pillow==8.1.0
+# for testing
+requests==2.25.1