diff --git a/modoboa/admin/management/commands/subcommands/_import.py b/modoboa/admin/management/commands/subcommands/_import.py index 0c607b604..e3da57c0a 100644 --- a/modoboa/admin/management/commands/subcommands/_import.py +++ b/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: diff --git a/modoboa/admin/tests/test_data/import_domains.csv b/modoboa/admin/tests/test_data/import_domains.csv new file mode 100644 index 000000000..d879f8a16 --- /dev/null +++ b/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 diff --git a/modoboa/admin/tests/test_data/import_domains_duplicates.csv b/modoboa/admin/tests/test_data/import_domains_duplicates.csv new file mode 100644 index 000000000..c5395f364 --- /dev/null +++ b/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 diff --git a/modoboa/admin/tests/test_import_.py b/modoboa/admin/tests/test_import_.py index 540fede38..c7ec8e1ff 100644 --- a/modoboa/admin/tests/test_import_.py +++ b/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))