fix import command

- ensure command line args are unicode(py2)/str(py3)

- use io.open to open file with utf8 encoding

- set newlines="" in open commands, csv module takes care of line endings

- add tests for import command

modoboa/admin/management/commands/subcommands/_import.py

@@ -11,6 +11,7 @@ import progressbar
 from backports import csv
 
 from django.core.management.base import BaseCommand, CommandError
+from django.utils import six
 
 from modoboa.core import models as core_models
 from modoboa.core.extensions import exts_pool
@@ -27,7 +28,7 @@ class ImportCommand(BaseCommand):
     def add_arguments(self, parser):
         """Add extra arguments to command."""
         parser.add_argument(
-            "--sepchar", type=str, default=";",
+            "--sepchar", type=six.text_type, default=";",
             help="Separator used in file.")
         parser.add_argument(
             "--continue-if-exists", action="store_true",
@@ -37,23 +38,25 @@ class ImportCommand(BaseCommand):
             "--crypt-password", action="store_true",
             default=False, help="Encrypt provided passwords.")
         parser.add_argument(
-            "files", type=str, nargs="+", help="CSV files to import.")
+            "files", type=six.text_type, nargs="+", help="CSV files to import.")
 
     def _import(self, filename, options):
         """Import domains or identities."""
         superadmin = (
             core_models.User.objects.filter(is_superuser=True).first()
         )
-        if not os.path.exists(filename):
+        if not os.path.isfile(filename):
             raise CommandError("File not found")
 
-        num_lines = sum(1 for line in open(filename) if line)
+        num_lines = sum(
+            1 for line in io.open(filename, encoding="utf-8") if line
+        )
         pbar = progressbar.ProgressBar(
             widgets=[
                 progressbar.Percentage(), progressbar.Bar(), progressbar.ETA()
             ], maxval=num_lines
         ).start()
-        with io.open(filename, encoding="utf8") as f:
+        with io.open(filename, encoding="utf-8", newline="") as f:
             reader = csv.reader(f, delimiter=options["sepchar"])
             i = 0
             for row in reader:

modoboa/admin/tests/test_data/import_domains.csv

@@ -0,0 +1,2 @@
+domain;domain1.com;1000;100;True
+domain;dómain2.com;2000;200;True

modoboa/admin/tests/test_data/import_domains_duplicates.csv

@@ -0,0 +1,3 @@
+domain;domain1.com;1000;100;True
+domain;dómain2.com;2000;200;True
+domain;dómain2.com;2000;200;True

modoboa/admin/tests/test_import_.py

@@ -2,10 +2,15 @@
 
 from __future__ import unicode_literals
 
+import os
+
 import dns.resolver
+import mock
 from mock import patch
 
 from django.core.files.base import ContentFile
+from django.core.management import call_command
+from django.core.management.base import CommandError
 from django.urls import reverse
 
 from modoboa.core import factories as core_factories
@@ -347,3 +352,63 @@ dómain; dómªin2.com; 1000; 100; True
         self.assertEqual(dom.default_mailbox_quota, 100)
         self.assertTrue(dom.enabled)
         self.assertTrue(admin.is_owner(dom))
+
+    def test_import_command_missing_file(self):
+        with mock.patch("os.path.isfile") as mock_isfile:
+            mock_isfile.return_value = False
+            with self.assertRaises(CommandError) as cm:
+                call_command("modo", "import", "test.csv")
+                ex_message = cm.exception.messages
+                self.assertEqual(ex_message, "File not found")
+
+    def test_import_command(self):
+        test_file = os.path.join(
+            os.path.dirname(__file__),
+            "test_data/import_domains.csv"
+        )
+        call_command("modo", "import", test_file)
+
+        admin = User.objects.get(username="admin")
+
+        dom = Domain.objects.get(name="domain1.com")
+        self.assertEqual(dom.quota, 1000)
+        self.assertEqual(dom.default_mailbox_quota, 100)
+        self.assertTrue(dom.enabled)
+        self.assertTrue(admin.is_owner(dom))
+
+        dom = Domain.objects.get(name="dómain2.com")
+        self.assertEqual(dom.quota, 2000)
+        self.assertEqual(dom.default_mailbox_quota, 200)
+        self.assertTrue(dom.enabled)
+        self.assertTrue(admin.is_owner(dom))
+
+    def test_import_command_duplicates(self):
+        test_file = os.path.join(
+            os.path.dirname(__file__),
+            "test_data/import_domains_duplicates.csv"
+        )
+        with self.assertRaises(CommandError) as cm:
+            call_command("modo", "import", test_file)
+            ex_message = cm.exception.messages
+            self.assertTrue(ex_message.startswith("Object already exists: "))
+
+    def test_import_command_duplicates_continue(self):
+        test_file = os.path.join(
+            os.path.dirname(__file__),
+            "test_data/import_domains_duplicates.csv"
+        )
+        call_command("modo", "import", "--continue-if-exists", test_file)
+
+        admin = User.objects.get(username="admin")
+
+        dom = Domain.objects.get(name="domain1.com")
+        self.assertEqual(dom.quota, 1000)
+        self.assertEqual(dom.default_mailbox_quota, 100)
+        self.assertTrue(dom.enabled)
+        self.assertTrue(admin.is_owner(dom))
+
+        dom = Domain.objects.get(name="dómain2.com")
+        self.assertEqual(dom.quota, 2000)
+        self.assertEqual(dom.default_mailbox_quota, 200)
+        self.assertTrue(dom.enabled)
+        self.assertTrue(admin.is_owner(dom))