added test for product load

products/tests.py

@@ -259,3 +259,78 @@ class ProductViewSetTest(APITestCase):
         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())
+
+
+class LoadCoopProductsTestCase(APITestCase):
+
+    def setUp(self):
+        """Tests setup
+        """
+        self.endpoint = '/api/v1/load_products/'
+        self.model = Product
+        self.factory = ProductFactory
+        # create admin user
+        self.admin_email = f"admin_user@mail.com"
+        self.password = ''.join(random.choices(string.ascii_uppercase, k = 10))
+        self.admin_user = CustomUserFactory(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 = CustomUserFactory(email=self.reg_email, is_active=True)
+        self.user.set_password(self.password)
+        self.user.save()
+        # test CSV file path
+        self.csv_path = 'datasets/test_products.csv'
+
+    def test_admin_can_load_csv(self):
+        # delete existing instances
+        self.model.objects.all().delete()
+
+        # read csv file
+        files = {'csv_file': open(self.csv_path,'rt')}
+
+        # Authenticate
+        token = get_tokens_for_user(self.admin_user)
+        self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
+
+        # send in request
+        response = self.client.post(self.endpoint, files)
+
+        # check re sponse
+        self.assertEqual(response.status_code, 200)
+        # check for object creation
+        self.assertEquals(5, self.model.objects.count())
+
+    def test_auth_user_cannot_load_csv(self):
+        # delete existing instances
+        self.model.objects.all().delete()
+
+        # read csv file
+        files = {'csv_file': open(self.csv_path,'r')}
+
+        # Authenticate
+        token = get_tokens_for_user(self.user)
+        self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
+
+        # send in request
+        response = self.client.post(self.endpoint, files)
+
+        # check response
+        self.assertEqual(response.status_code, 403)
+        # check for object creation
+        self.assertEqual(0, self.model.objects.count())
+
+    def test_anon_user_cannot_load_csv(self):
+        # delete existing instances
+        self.model.objects.all().delete()
+
+        # read csv file
+        files = {'csv_file': open(self.csv_path,'r')}
+
+        # send in request
+        response = self.client.post(self.endpoint, files)
+
+        # check response
+        self.assertEqual(response.status_code, 401)
+        # check for object creation
+        self.assertEqual(0, self.model.objects.count())
+

products/views.py

@@ -6,6 +6,7 @@ from django.conf import settings
 
 # Create your views here.
 from rest_framework import viewsets
+from rest_framework.response import Response
 from rest_framework.permissions import IsAuthenticatedOrReadOnly, IsAdminUser
 from rest_framework.decorators import api_view, permission_classes