[utils] Add error handling in csvfile validation #251

Fixes #251 1.Add try except blocks to catch the error 2.Add test for the same 3.Use ugettext_lazy 4.Fix Formatting in utils file
openwisp · Nov 20, 2019 · 94845ed · 94845ed
1 parent 3b745b5
commit 94845ed
Show file tree

Hide file tree

Showing 3 changed files with 37 additions and 22 deletions.
diff --git a/django_freeradius/tests/base/test_utils.py b/django_freeradius/tests/base/test_utils.py
@@ -7,17 +7,24 @@
 class BaseTestUtils(object):
     def test_find_available_username(self):
         User = get_user_model()
-        User.objects.create(username='rohith', password='password')
-        self.assertEqual(find_available_username('rohith', []), 'rohith1')
-        User.objects.create(username='rohith1', password='password')
-        self.assertEqual(find_available_username('rohith', []), 'rohith2')
+        User.objects.create(username="rohith", password="password")
+        self.assertEqual(find_available_username("rohith", []), "rohith1")
+        User.objects.create(username="rohith1", password="password")
+        self.assertEqual(find_available_username("rohith", []), "rohith2")
 
     def test_validate_csvfile(self):
-        invalid_csv_path = self._get_path('static/test_batch_invalid.csv')
-        improper_csv_path = self._get_path('static/test_batch_improper.csv')
+        invalid_csv_path = self._get_path("static/test_batch_invalid.csv")
+        improper_csv_path = self._get_path("static/test_batch_improper.csv")
+        invalid_format_path = self._get_path("static/test_batch_invalid_format.pdf")
         with self.assertRaises(ValidationError) as error:
-            validate_csvfile(open(invalid_csv_path, 'rt'))
-        self.assertTrue('Enter a valid email address' in error.exception.message)
+            validate_csvfile(open(invalid_format_path, "rb"))
+        self.assertTrue(
+            "Incorrect file format has been uploaded. Please Try Again!"
+            in error.exception.message
+        )
         with self.assertRaises(ValidationError) as error:
-            validate_csvfile(open(improper_csv_path, 'rt'))
-        self.assertTrue('Improper CSV format' in error.exception.message)
+            validate_csvfile(open(invalid_csv_path, "rt"))
+        self.assertTrue("Enter a valid email address" in error.exception.message)
+        with self.assertRaises(ValidationError) as error:
+            validate_csvfile(open(improper_csv_path, "rt"))
+        self.assertTrue("Improper CSV format" in error.exception.message)
diff --git a/django_freeradius/tests/static/test_batch_invalid_format.pdf b/django_freeradius/tests/static/test_batch_invalid_format.pdf
diff --git a/django_freeradius/utils.py b/django_freeradius/utils.py
@@ -18,19 +18,24 @@
 def find_available_username(username, users_list, prefix=False):
     User = get_user_model()
     suffix = 1
-    tmp = '{}{}'.format(username, suffix) if prefix else username
+    tmp = "{}{}".format(username, suffix) if prefix else username
     names_list = map(lambda x: x.username, users_list)
     while User.objects.filter(username=tmp).exists() or tmp in names_list:
         suffix += 1 if prefix else 0
-        tmp = '{}{}'.format(username, suffix)
+        tmp = "{}{}".format(username, suffix)
         suffix += 1 if not prefix else 0
     return tmp
 
 
 def validate_csvfile(csvfile):
     csv_data = csvfile.read()
-    csv_data = csv_data.decode('utf-8') if isinstance(csv_data, bytes) else csv_data
-    reader = csv.reader(StringIO(csv_data), delimiter=',')
+    try:
+        csv_data = csv_data.decode("utf-8") if isinstance(csv_data, bytes) else csv_data
+    except UnicodeDecodeError:
+        raise ValidationError(
+            _("Incorrect file format has been uploaded. Please Try Again!")
+        )
+    reader = csv.reader(StringIO(csv_data), delimiter=",")
     error_message = "The CSV contains a line with invalid data,\
                     line number {} triggered the following error: {}"
     row_count = 1
@@ -40,10 +45,14 @@ def validate_csvfile(csvfile):
             try:
                 validate_email(email)
             except ValidationError as e:
-                raise ValidationError(_(error_message.format(str(row_count), e.message)))
+                raise ValidationError(
+                    _(error_message.format(str(row_count), e.message))
+                )
             row_count += 1
         elif len(row) > 0:
-            raise ValidationError(_(error_message.format(str(row_count), "Improper CSV format.")))
+            raise ValidationError(
+                _(error_message.format(str(row_count), "Improper CSV format."))
+            )
     csvfile.seek(0)
 
 
@@ -64,20 +73,19 @@ def prefix_generate_users(prefix, n, password_length):
 def generate_pdf(prefix, data):
     template = get_template(BATCH_PDF_TEMPLATE)
     html = template.render(data)
-    f = open('{}/{}.pdf'.format(settings.MEDIA_ROOT, prefix), 'w+b')
-    pisa.CreatePDF(html.encode('utf-8'), dest=f, encoding='utf-8')
+    f = open("{}/{}.pdf".format(settings.MEDIA_ROOT, prefix), "w+b")
+    pisa.CreatePDF(html.encode("utf-8"), dest=f, encoding="utf-8")
     f.seek(0)
     return File(f)
 
 
 def set_default_group(sender, instance, created, **kwargs):
     if created:
-        RadiusGroup = swapper.load_model('django_freeradius', 'RadiusGroup')
-        RadiusUserGroup = swapper.load_model('django_freeradius', 'RadiusUserGroup')
+        RadiusGroup = swapper.load_model("django_freeradius", "RadiusGroup")
+        RadiusUserGroup = swapper.load_model("django_freeradius", "RadiusUserGroup")
         queryset = RadiusGroup.objects.filter(default=True)
         if queryset.exists():
-            ug = RadiusUserGroup(user=instance,
-                                 group=queryset.first())
+            ug = RadiusUserGroup(user=instance, group=queryset.first())
             ug.full_clean()
             ug.save()