Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Split test_resources.py into test_resources package
- Loading branch information
Showing
20 changed files
with
2,091 additions
and
1,888 deletions.
There are no files selected for viewing
Empty file.
155 changes: 155 additions & 0 deletions
155
tests/core/tests/test_resources/test_modelresource/test_data_deletion_dry_run.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,155 @@ | ||
from unittest import mock | ||
|
||
import tablib | ||
from core.models import Book | ||
from core.tests.resources import BookResource | ||
from core.tests.utils import ignore_widget_deprecation_warning | ||
from django.test import TestCase | ||
|
||
from import_export import fields, results, widgets | ||
|
||
|
||
class DataDeletionDryRunTests(TestCase): | ||
def setUp(self): | ||
self.resource = BookResource() | ||
self.book = Book.objects.create(name="Some book") | ||
self.dataset = tablib.Dataset(headers=["id", "name", "author_email", "price"]) | ||
row = [self.book.pk, "Some book", "test@example.com", "10.25"] | ||
self.dataset.append(row) | ||
|
||
@ignore_widget_deprecation_warning | ||
def test_import_data_delete(self): | ||
class B(BookResource): | ||
delete = fields.Field(widget=widgets.BooleanWidget()) | ||
|
||
def for_delete(self, row, instance): | ||
return self.fields["delete"].clean(row) | ||
|
||
row = [self.book.pk, self.book.name, "1"] | ||
dataset = tablib.Dataset(*[row], headers=["id", "name", "delete"]) | ||
result = B().import_data(dataset, raise_errors=True) | ||
self.assertFalse(result.has_errors()) | ||
self.assertEqual( | ||
result.rows[0].import_type, results.RowResult.IMPORT_TYPE_DELETE | ||
) | ||
self.assertFalse(Book.objects.filter(pk=self.book.pk)) | ||
self.assertIsNone(result.rows[0].instance) | ||
self.assertIsNone(result.rows[0].original) | ||
|
||
@ignore_widget_deprecation_warning | ||
def test_import_data_delete_store_instance(self): | ||
class B(BookResource): | ||
delete = fields.Field(widget=widgets.BooleanWidget()) | ||
|
||
def for_delete(self, row, instance): | ||
return self.fields["delete"].clean(row) | ||
|
||
class Meta: | ||
store_instance = True | ||
|
||
row = [self.book.pk, self.book.name, "1"] | ||
dataset = tablib.Dataset(*[row], headers=["id", "name", "delete"]) | ||
result = B().import_data(dataset, raise_errors=True) | ||
self.assertEqual( | ||
result.rows[0].import_type, results.RowResult.IMPORT_TYPE_DELETE | ||
) | ||
self.assertIsNotNone(result.rows[0].instance) | ||
|
||
@ignore_widget_deprecation_warning | ||
def test_save_instance_with_dry_run_flag(self): | ||
class B(BookResource): | ||
def before_save_instance(self, instance, row, **kwargs): | ||
super().before_save_instance(instance, row, **kwargs) | ||
dry_run = kwargs.get("dry_run", False) | ||
if dry_run: | ||
self.before_save_instance_dry_run = True | ||
else: | ||
self.before_save_instance_dry_run = False | ||
|
||
def save_instance(self, instance, new, row, **kwargs): | ||
super().save_instance(instance, new, row, **kwargs) | ||
dry_run = kwargs.get("dry_run", False) | ||
if dry_run: | ||
self.save_instance_dry_run = True | ||
else: | ||
self.save_instance_dry_run = False | ||
|
||
def after_save_instance(self, instance, row, **kwargs): | ||
super().after_save_instance(instance, row, **kwargs) | ||
dry_run = kwargs.get("dry_run", False) | ||
if dry_run: | ||
self.after_save_instance_dry_run = True | ||
else: | ||
self.after_save_instance_dry_run = False | ||
|
||
resource = B() | ||
resource.import_data(self.dataset, dry_run=True, raise_errors=True) | ||
self.assertTrue(resource.before_save_instance_dry_run) | ||
self.assertTrue(resource.save_instance_dry_run) | ||
self.assertTrue(resource.after_save_instance_dry_run) | ||
|
||
resource.import_data(self.dataset, dry_run=False, raise_errors=True) | ||
self.assertFalse(resource.before_save_instance_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, False, None, 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, None, using_transactions=False, dry_run=True | ||
) | ||
self.assertEqual(0, mock_book.call_count) | ||
|
||
@ignore_widget_deprecation_warning | ||
def test_delete_instance_with_dry_run_flag(self): | ||
class B(BookResource): | ||
delete = fields.Field(widget=widgets.BooleanWidget()) | ||
|
||
def for_delete(self, row, instance): | ||
return self.fields["delete"].clean(row) | ||
|
||
def before_delete_instance(self, instance, row, **kwargs): | ||
super().before_delete_instance(instance, row, **kwargs) | ||
dry_run = kwargs.get("dry_run", False) | ||
if dry_run: | ||
self.before_delete_instance_dry_run = True | ||
else: | ||
self.before_delete_instance_dry_run = False | ||
|
||
def delete_instance(self, instance, row, **kwargs): | ||
super().delete_instance(instance, row, **kwargs) | ||
dry_run = kwargs.get("dry_run", False) | ||
if dry_run: | ||
self.delete_instance_dry_run = True | ||
else: | ||
self.delete_instance_dry_run = False | ||
|
||
def after_delete_instance(self, instance, row, **kwargs): | ||
super().after_delete_instance(instance, row, **kwargs) | ||
dry_run = kwargs.get("dry_run", False) | ||
if dry_run: | ||
self.after_delete_instance_dry_run = True | ||
else: | ||
self.after_delete_instance_dry_run = False | ||
|
||
resource = B() | ||
row = [self.book.pk, self.book.name, "1"] | ||
dataset = tablib.Dataset(*[row], headers=["id", "name", "delete"]) | ||
resource.import_data(dataset, dry_run=True, raise_errors=True) | ||
self.assertTrue(resource.before_delete_instance_dry_run) | ||
self.assertTrue(resource.delete_instance_dry_run) | ||
self.assertTrue(resource.after_delete_instance_dry_run) | ||
|
||
resource.import_data(dataset, dry_run=False, raise_errors=True) | ||
self.assertFalse(resource.before_delete_instance_dry_run) | ||
self.assertFalse(resource.delete_instance_dry_run) | ||
self.assertFalse(resource.after_delete_instance_dry_run) |
121 changes: 121 additions & 0 deletions
121
tests/core/tests/test_resources/test_modelresource/test_data_handling.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,121 @@ | ||
from decimal import InvalidOperation | ||
|
||
import tablib | ||
from core.models import Author, Book | ||
from core.tests.resources import AuthorResourceWithCustomWidget, BookResource | ||
from core.tests.utils import ignore_widget_deprecation_warning | ||
from django.test import TestCase | ||
|
||
from import_export import resources, results | ||
|
||
|
||
class DataHandlingTests(TestCase): | ||
def setUp(self): | ||
self.resource = BookResource() | ||
self.book = Book.objects.create(name="Some book") | ||
self.dataset = tablib.Dataset(headers=["id", "name", "author_email", "price"]) | ||
row = [self.book.pk, "Some book", "test@example.com", "10.25"] | ||
self.dataset.append(row) | ||
|
||
@ignore_widget_deprecation_warning | ||
def test_import_data_handles_widget_valueerrors_with_unicode_messages(self): | ||
resource = AuthorResourceWithCustomWidget() | ||
dataset = tablib.Dataset(headers=["id", "name", "birthday"]) | ||
dataset.append(["", "A.A.Milne", "1882-01-18"]) | ||
|
||
result = resource.import_data(dataset, raise_errors=False) | ||
|
||
self.assertTrue(result.has_validation_errors()) | ||
self.assertIs(result.rows[0].import_type, results.RowResult.IMPORT_TYPE_INVALID) | ||
self.assertEqual( | ||
result.invalid_rows[0].field_specific_errors["name"], | ||
["Ова вриједност је страшна!"], | ||
) | ||
|
||
def test_model_validation_errors_not_raised_when_clean_model_instances_is_false( | ||
self, | ||
): | ||
class TestResource(resources.ModelResource): | ||
class Meta: | ||
model = Author | ||
clean_model_instances = False | ||
|
||
resource = TestResource() | ||
dataset = tablib.Dataset(headers=["id", "name"]) | ||
dataset.append(["", "123"]) | ||
|
||
result = resource.import_data(dataset, raise_errors=False) | ||
self.assertFalse(result.has_validation_errors()) | ||
self.assertEqual(len(result.invalid_rows), 0) | ||
|
||
@ignore_widget_deprecation_warning | ||
def test_model_validation_errors_raised_when_clean_model_instances_is_true(self): | ||
class TestResource(resources.ModelResource): | ||
class Meta: | ||
model = Author | ||
clean_model_instances = True | ||
export_order = ["id", "name", "birthday"] | ||
|
||
# create test dataset | ||
# NOTE: column order is deliberately strange | ||
dataset = tablib.Dataset(headers=["name", "id"]) | ||
dataset.append(["123", "1"]) | ||
|
||
# run import_data() | ||
resource = TestResource() | ||
result = resource.import_data(dataset, raise_errors=False) | ||
|
||
# check has_validation_errors() | ||
self.assertTrue(result.has_validation_errors()) | ||
|
||
# check the invalid row itself | ||
invalid_row = result.invalid_rows[0] | ||
self.assertEqual(invalid_row.error_count, 1) | ||
self.assertEqual( | ||
invalid_row.field_specific_errors, {"name": ["'123' is not a valid value"]} | ||
) | ||
# diff_header and invalid_row.values should match too | ||
self.assertEqual(result.diff_headers, ["id", "name", "birthday"]) | ||
self.assertEqual(invalid_row.values, ("1", "123", "---")) | ||
|
||
@ignore_widget_deprecation_warning | ||
def test_known_invalid_fields_are_excluded_from_model_instance_cleaning(self): | ||
# The custom widget on the parent class should complain about | ||
# 'name' first, preventing Author.full_clean() from raising the | ||
# error as it does in the previous test | ||
|
||
class TestResource(AuthorResourceWithCustomWidget): | ||
class Meta: | ||
model = Author | ||
clean_model_instances = True | ||
|
||
resource = TestResource() | ||
dataset = tablib.Dataset(headers=["id", "name"]) | ||
dataset.append(["", "123"]) | ||
|
||
result = resource.import_data(dataset, raise_errors=False) | ||
self.assertTrue(result.has_validation_errors()) | ||
self.assertEqual(result.invalid_rows[0].error_count, 1) | ||
self.assertEqual( | ||
result.invalid_rows[0].field_specific_errors, | ||
{"name": ["Ова вриједност је страшна!"]}, | ||
) | ||
|
||
def test_import_data_error_saving_model(self): | ||
row = list(self.dataset.pop()) | ||
# set pk to something that would yield error | ||
row[0] = "foo" | ||
self.dataset.append(row) | ||
result = self.resource.import_data(self.dataset, raise_errors=False) | ||
|
||
self.assertTrue(result.has_errors()) | ||
self.assertTrue(result.rows[0].errors) | ||
actual = result.rows[0].errors[0].error | ||
self.assertIsInstance(actual, (ValueError, InvalidOperation)) | ||
self.assertIn( | ||
str(actual), | ||
{ | ||
"could not convert string to float", | ||
"[<class 'decimal.ConversionSyntax'>]", | ||
}, | ||
) |
Oops, something went wrong.