increased test coverage

Makefile

@@ -23,7 +23,7 @@ clean-tests: ## remove pytest artifacts
 	rm -fr django-import-export/
 
 lint: ## check style with isort
-	isort --check-only
+	isort --check-only .
 
 test: ## run tests quickly with the default Python
 	$(RUN_TEST_COMMAND)

tests/core/tests/test_base_formats.py

@@ -1,6 +1,7 @@
 import os
 from unittest import mock
 
+import tablib
 from django.test import TestCase
 from django.utils.encoding import force_str
 from tablib.core import UnsupportedFormat
@@ -10,6 +11,9 @@
 
 class FormatTest(TestCase):
 
+    def setUp(self):
+        self.format = base_formats.Format()
+
     @mock.patch('import_export.formats.base_formats.HTML.get_format', side_effect=ImportError)
     def test_format_non_available1(self, mocked):
         self.assertFalse(base_formats.HTML.is_available())
@@ -21,11 +25,48 @@ def test_format_non_available2(self, mocked):
     def test_format_available(self):
         self.assertTrue(base_formats.CSV.is_available())
 
+    def test_get_title(self):
+        self.assertEqual("<class 'import_export.formats.base_formats.Format'>", str(self.format.get_title()))
+
+    def test_create_dataset_NotImplementedError(self):
+        with self.assertRaises(NotImplementedError):
+            self.format.create_dataset(None)
+
+    def test_export_data_NotImplementedError(self):
+        with self.assertRaises(NotImplementedError):
+            self.format.export_data(None)
+
+    def test_get_extension(self):
+        self.assertEqual("", self.format.get_extension())
+
+    def test_get_content_type(self):
+        self.assertEqual("application/octet-stream", self.format.get_content_type())
+
+    def test_is_available_default(self):
+        self.assertTrue(self.format.is_available())
+
+    def test_can_import_default(self):
+        self.assertFalse(self.format.can_import())
+
+    def test_can_export_default(self):
+        self.assertFalse(self.format.can_export())
 
 class XLSTest(TestCase):
 
+    def setUp(self):
+        self.format = base_formats.XLS()
+
     def test_binary_format(self):
-        self.assertTrue(base_formats.XLS().is_binary())
+        self.assertTrue(self.format.is_binary())
+
+    def test_import(self):
+        filename = os.path.join(
+            os.path.dirname(__file__),
+            os.path.pardir,
+            'exports',
+            'books.xls')
+        with open(filename, self.format.get_read_mode()) as in_stream:
+            self.format.create_dataset(in_stream.read())
 
 
 class XLSXTest(TestCase):
@@ -50,6 +91,8 @@ class CSVTest(TestCase):
 
     def setUp(self):
         self.format = base_formats.CSV()
+        self.dataset = tablib.Dataset(headers=['id', 'username'])
+        self.dataset.append(('1', 'x'))
 
     def test_import_dos(self):
         filename = os.path.join(
@@ -84,6 +127,22 @@ def test_import_unicode(self):
             data = force_str(in_stream.read())
         base_formats.CSV().create_dataset(data)
 
+    def test_export_data(self):
+        res = self.format.export_data(self.dataset)
+        self.assertEqual("id,username\r\n1,x\r\n", res)
+
+    def test_get_extension(self):
+        self.assertEqual("csv", self.format.get_extension())
+
+    def test_content_type(self):
+        self.assertEqual("text/csv", self.format.get_content_type())
+
+    def test_can_import(self):
+        self.assertTrue(self.format.can_import())
+
+    def test_can_export(self):
+        self.assertTrue(self.format.can_export())
+
 
 class TSVTest(TestCase):
 
@@ -111,3 +170,15 @@ def test_import_unicode(self):
         with open(filename, self.format.get_read_mode()) as in_stream:
             data = force_str(in_stream.read())
         base_formats.TSV().create_dataset(data)
+
+
+class TextFormatTest(TestCase):
+
+    def setUp(self):
+        self.format = base_formats.TextFormat()
+
+    def test_get_read_mode(self):
+        self.assertEqual('r', self.format.get_read_mode())
+
+    def test_is_binary(self):
+        self.assertFalse(self.format.is_binary())

tests/core/tests/test_fields.py

@@ -25,6 +25,11 @@ def test_clean(self):
         self.assertEqual(self.field.clean(self.row),
                          self.row['name'])
 
+    def test_clean_raises_KeyError(self):
+        self.field.column_name = 'x'
+        with self.assertRaisesRegex(KeyError, "Column 'x' not found in dataset. Available columns are: \\['name'\\]"):
+            self.field.clean(self.row)
+
     def test_export(self):
         self.assertEqual(self.field.export(self.obj),
                          self.row['name'])

tests/core/tests/test_instance_loaders.py

@@ -5,6 +5,27 @@
 from import_export import instance_loaders, resources
 
 
+class BaseInstanceLoaderTest(TestCase):
+
+    def test_get_instance(self):
+        instance_loader = instance_loaders.BaseInstanceLoader(None)
+        with self.assertRaises(NotImplementedError):
+            instance_loader.get_instance(None)
+
+
+class ModelInstanceLoaderTest(TestCase):
+
+    def setUp(self):
+        self.resource = resources.modelresource_factory(Book)()
+
+    def test_get_instance_returns_None_when_params_is_empty(self):
+        # setting an empty array of import_id_fields will mean
+        # that 'params' is never set
+        self.resource._meta.import_id_fields = []
+        instance_loader = instance_loaders.ModelInstanceLoader(self.resource)
+        self.assertIsNone(instance_loader.get_instance([]))
+
+
 class CachedInstanceLoaderTest(TestCase):
 
     def setUp(self):
@@ -28,7 +49,6 @@ def test_get_instance(self):
         self.assertEqual(obj, self.book)
 
 
-
 class CachedInstanceLoaderWithAbsentImportIdFieldTest(TestCase):
     """Ensure that the cache is empty when the PK field is absent
     in the inbound dataset.

tests/core/tests/test_resources.py

@@ -9,7 +9,7 @@
 import tablib
 from django.conf import settings
 from django.contrib.auth.models import User
-from django.core.exceptions import ValidationError
+from django.core.exceptions import ImproperlyConfigured, ValidationError
 from django.core.paginator import Paginator
 from django.db import IntegrityError
 from django.db.models import Count
@@ -134,6 +134,22 @@ class B(A):
         resource = B()
         self.assertEqual(resource._meta.custom_attribute, True)
 
+    def test_get_use_transactions_defined_in_resource(self):
+        class A(MyResource):
+            class Meta:
+                use_transactions = True
+        resource = A()
+        self.assertTrue(resource.get_use_transactions())
+
+    def test_get_field_name_raises_AttributeError(self):
+        err = "Field x does not exists in <class 'core.tests.test_resources.MyResource'> resource"
+        with self.assertRaisesRegex(AttributeError, err):
+            self.my_resource.get_field_name('x')
+
+    def test_init_instance_raises_NotImplementedError(self):
+        with self.assertRaises(NotImplementedError):
+            self.my_resource.init_instance([])
+
 
 class AuthorResource(resources.ModelResource):
 
@@ -412,6 +428,21 @@ def test_import_data(self):
         self.assertEqual(instance.author_email, 'test@example.com')
         self.assertEqual(instance.price, Decimal("10.25"))
 
+    @mock.patch("import_export.resources.connections")
+    def test_raised_ImproperlyConfigured_if_use_transactions_set_when_transactions_not_supported(self, mock_db_connections):
+        class Features(object):
+            supports_transactions = False
+        class DummyConnection(object):
+            features = Features()
+
+        dummy_connection = DummyConnection()
+        mock_db_connections.__getitem__.return_value = dummy_connection
+        with self.assertRaises(ImproperlyConfigured):
+            self.resource.import_data(
+                self.dataset,
+                use_transactions=True,
+            )
+
     def test_importing_with_line_number_logging(self):
         resource = BookResourceWithLineNumberLogger()
         result = resource.import_data(self.dataset, raise_errors=True)
@@ -429,6 +460,63 @@ def test_import_data_raises_field_specific_validation_errors(self):
         self.assertIs(result.rows[0].import_type, results.RowResult.IMPORT_TYPE_INVALID)
         self.assertIn('birthday', result.invalid_rows[0].field_specific_errors)
 
+    def test_collect_failed_rows(self):
+        resource = ProfileResource()
+        headers = ['id', 'user']
+        # 'user' is a required field, the database will raise an error.
+        row = [None, None]
+        dataset = tablib.Dataset(row, headers=headers)
+        result = resource.import_data(
+            dataset, dry_run=True, use_transactions=True,
+            collect_failed_rows=True,
+        )
+        self.assertEqual(
+            result.failed_dataset.headers,
+            ['id', 'user', 'Error']
+        )
+        self.assertEqual(len(result.failed_dataset), 1)
+        # We can't check the error message because it's package- and version-dependent
+
+    def test_row_result_raise_errors(self):
+        resource = ProfileResource()
+        headers = ['id', 'user']
+        # 'user' is a required field, the database will raise an error.
+        row = [None, None]
+        dataset = tablib.Dataset(row, headers=headers)
+        with self.assertRaises(IntegrityError):
+            resource.import_data(
+                dataset, dry_run=True, use_transactions=True,
+                raise_errors=True,
+            )
+
+    def test_collect_failed_rows_validation_error(self):
+        resource = ProfileResource()
+        row = ['1']
+        dataset = tablib.Dataset(row, headers=['id'])
+        with mock.patch("import_export.resources.Field.save", side_effect=ValidationError("fail!")):
+            result = resource.import_data(
+                dataset, dry_run=True, use_transactions=True,
+                collect_failed_rows=True,
+            )
+        self.assertEqual(
+            result.failed_dataset.headers,
+            ['id', 'Error']
+        )
+        self.assertEqual(1, len(result.failed_dataset), )
+        self.assertEqual('1', result.failed_dataset.dict[0]['id'])
+        self.assertEqual("{'__all__': ['fail!']}", result.failed_dataset.dict[0]['Error'])
+
+    def test_row_result_raise_ValidationError(self):
+        resource = ProfileResource()
+        row = ['1']
+        dataset = tablib.Dataset(row, headers=['id'])
+        with mock.patch("import_export.resources.Field.save", side_effect=ValidationError("fail!")):
+            with self.assertRaisesRegex(ValidationError, "{'__all__': \\['fail!'\\]}") :
+                resource.import_data(
+                    dataset, dry_run=True, use_transactions=True,
+                    raise_errors=True,
+                )
+
     def test_import_data_handles_widget_valueerrors_with_unicode_messages(self):
         resource = AuthorResourceWithCustomWidget()
         dataset = tablib.Dataset(headers=['id', 'name', 'birthday'])
@@ -579,6 +667,18 @@ def after_save_instance(self, instance, using_transactions, dry_run):
         self.assertFalse(resource.save_instance_dry_run)
         self.assertFalse(resource.after_save_instance_dry_run)
 
+    @mock.patch("core.models.Book.save")
+    def test_save_instance_noop(self, mock_book):
+        book = Book.objects.first()
+        self.resource.save_instance(book, using_transactions=False, dry_run=True)
+        self.assertEqual(0, mock_book.call_count)
+
+    @mock.patch("core.models.Book.save")
+    def test_delete_instance_noop(self, mock_book):
+        book = Book.objects.first()
+        self.resource.delete_instance(book, using_transactions=False, dry_run=True)
+        self.assertEqual(0, mock_book.call_count)
+
     def test_delete_instance_with_dry_run_flag(self):
         class B(BookResource):
             delete = fields.Field(widget=widgets.BooleanWidget())
@@ -1085,24 +1185,6 @@ def test_create_object_after_importing_dataset_with_id(self):
         except IntegrityError:
             self.fail('IntegrityError was raised.')
 
-    def test_collect_failed_rows(self):
-        resource = ProfileResource()
-        headers = ['id', 'user']
-        # 'user' is a required field, the database will raise an error.
-        row = [None, None]
-        dataset = tablib.Dataset(row, headers=headers)
-        result = resource.import_data(
-            dataset, dry_run=True, use_transactions=True,
-            collect_failed_rows=True,
-        )
-        self.assertEqual(
-            result.failed_dataset.headers,
-            ['id', 'user', 'Error']
-        )
-        self.assertEqual(len(result.failed_dataset), 1)
-        # We can't check the error message because it's package- and version-dependent
-
-
 if 'postgresql' in settings.DATABASES['default']['ENGINE']:
     from django.contrib.postgres.fields import ArrayField
     from django.db import models
@@ -1283,7 +1365,7 @@ class Meta:
         self.dataset.append(['', 'A.A.Milne', '1882test-01-18'])
 
     def test_skip_diff(self, mock_diff):
-        with mock.patch("copy.deepcopy") as mock_deep_copy:
+        with mock.patch("import_export.resources.deepcopy") as mock_deep_copy:
             self.resource.import_data(self.dataset)
             mock_diff.return_value.compare_with.assert_not_called()
             mock_diff.return_value.as_html.assert_not_called()
@@ -1300,7 +1382,7 @@ def for_delete(self, row, instance):
                 return True
 
         resource = BookResource()
-        with mock.patch("copy.deepcopy") as mock_deep_copy:
+        with mock.patch("import_export.resources.deepcopy") as mock_deep_copy:
             resource.import_data(self.dataset)
             mock_diff.return_value.compare_with.assert_not_called()
             mock_diff.return_value.as_html.assert_not_called()
@@ -1322,12 +1404,29 @@ def for_delete(self, row, instance):
 
         resource = BookResource()
 
-        with mock.patch("copy.deepcopy") as mock_deep_copy:
+        with mock.patch("import_export.resources.deepcopy") as mock_deep_copy:
             resource.import_data(self.dataset, dry_run=True)
             mock_diff.return_value.compare_with.assert_not_called()
             mock_diff.return_value.as_html.assert_not_called()
             mock_deep_copy.assert_not_called()
 
+    def test_skip_diff_for_delete_skip_row_not_enabled_new_object(self, mock_diff):
+        class BookResource(resources.ModelResource):
+
+            class Meta:
+                model = Book
+                skip_diff = False
+
+            def for_delete(self, row, instance):
+                return True
+
+        resource = BookResource()
+
+        with mock.patch("import_export.resources.deepcopy") as mock_deep_copy:
+            resource.import_data(self.dataset, dry_run=True)
+            self.assertEqual(1, mock_diff.return_value.compare_with.call_count)
+            self.assertEqual(1, mock_deep_copy.call_count)
+
     def test_skip_row_returns_false_when_skip_diff_is_true(self, mock_diff):
         class BookResource(resources.ModelResource):