| @@ -174,11 +174,38 @@ | ||
|---|---|---|
| + |
{% if row.import_type == 'new' %}
{% trans "New" %}
{% elif row.import_type == 'skip' %}
{% trans "Skipped" %}
+ {{ row.skipped_errors_count}}
+
+
{% elif row.import_type == 'delete' %}
{% trans "Delete" %}
{% elif row.import_type == 'update' %}
diff --git a/tests/conftest.py b/tests/conftest.py
index 7189b8f..9d5c04b 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,7 +1,10 @@
from django.conf import settings
from django.contrib.auth import get_user_model
+from django.contrib.auth.models import User
from django.core.files.uploadedfile import SimpleUploadedFile
+from rest_framework import test
+
import pytest
from tests.fake_app.models import Artist, Band, Membership
@@ -61,7 +64,7 @@ def membership(band: Band) -> Membership:
@pytest.fixture
-def uploaded_file(existing_artist: Artist):
+def uploaded_file(existing_artist: Artist) -> SimpleUploadedFile:
"""Generate valid `Artist` import file."""
import_job = ArtistImportJobFactory.build(artists=[existing_artist])
return SimpleUploadedFile(
@@ -71,6 +74,31 @@ def uploaded_file(existing_artist: Artist):
)
+@pytest.fixture
+def force_import_artist_job(new_artist: Artist) -> Artist:
+ """Return `ImportJob` with `force_import=True` and file with invalid row."""
+ return ArtistImportJobFactory(
+ artists=[new_artist],
+ is_valid_file=False,
+ force_import=True,
+ )
+
+
+@pytest.fixture
+def invalid_uploaded_file(new_artist: Artist) -> SimpleUploadedFile:
+ """Generate invalid `Artist` imort file."""
+ import_job = ArtistImportJobFactory.build(
+ artists=[new_artist],
+ force_import=True,
+ is_valid_file=False,
+ )
+ return SimpleUploadedFile(
+ "test_file.csv",
+ content=import_job.data_file.file.read().encode(),
+ content_type="text/plain",
+ )
+
+
@pytest.fixture
def superuser():
"""Return superuser instance."""
@@ -102,3 +130,19 @@ def temp_directory_for_media(tmpdir_factory):
)
media = tmpdir_factory.mktemp("tmp_media")
settings.MEDIA_ROOT = media
+
+
+@pytest.fixture
+def api_client() -> test.APIClient:
+ """Create api client."""
+ return test.APIClient()
+
+
+@pytest.fixture
+def admin_api_client(
+ superuser: User,
+ api_client: test.APIClient,
+) -> test.APIClient:
+ """Authenticate admin_user and return api client."""
+ api_client.force_authenticate(user=superuser)
+ return api_client
diff --git a/tests/fake_app/factories.py b/tests/fake_app/factories.py
index cfc6264..79317bf 100644
--- a/tests/fake_app/factories.py
+++ b/tests/fake_app/factories.py
@@ -63,11 +63,20 @@ class Meta:
class Params:
artists: list[models.Artist] = []
+ is_valid_file: bool = True
@factory.lazy_attribute
def data_file(self):
"""Generate `data_file` based on passed `artists`."""
resource = SimpleArtistResource()
+
+ if not self.is_valid_file:
+ # Append not existing artist with a non-existent instrument
+ # to violate the not-null constraint
+ self.artists.append(
+ ArtistFactory.build(instrument=InstrumentFactory.build()),
+ )
+
dataset = resource.export(self.artists)
export_data = formats.CSV().export_data(dataset)
content = django_files.ContentFile(export_data)
diff --git a/tests/settings.py b/tests/settings.py
index 3e3aa10..8a26705 100644
--- a/tests/settings.py
+++ b/tests/settings.py
@@ -9,6 +9,8 @@
# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = "a87082n4v52u4rnvk2edv128eudfvn5"
+ALLOWED_HOSTS = ["*"]
+
# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = True
diff --git a/tests/test_api/test_import.py b/tests/test_api/test_import.py
index f48ba00..59df495 100644
--- a/tests/test_api/test_import.py
+++ b/tests/test_api/test_import.py
@@ -1,9 +1,9 @@
from django.contrib.auth.models import User
from django.core.files.uploadedfile import SimpleUploadedFile
-from django.test.client import Client
from django.urls import reverse
from rest_framework import status
+from rest_framework.test import APIClient
import pytest
@@ -12,13 +12,11 @@
@pytest.mark.django_db(transaction=True)
def test_import_api_creates_import_job(
- client: Client,
- superuser: User,
+ admin_api_client: APIClient,
uploaded_file: SimpleUploadedFile,
):
"""Ensure import start api creates new import job."""
- client.force_login(superuser)
- response = client.post(
+ response = admin_api_client.post(
path=reverse("import-artist-start"),
data={
"file": uploaded_file,
@@ -32,27 +30,66 @@ def test_import_api_creates_import_job(
@pytest.mark.django_db(transaction=True)
def test_import_api_detail(
- client: Client,
- superuser: User,
+ admin_api_client: APIClient,
artist_import_job: ImportJob,
):
"""Ensure import detail api shows current import job status."""
- client.force_login(superuser)
- response = client.get(
+ response = admin_api_client.get(
path=reverse(
"import-artist-detail",
kwargs={"pk": artist_import_job.id},
),
)
+ assert response.status_code == status.HTTP_200_OK, response.data
assert response.data["import_status"] == ImportJob.ImportStatus.PARSED
artist_import_job.refresh_from_db()
artist_import_job.confirm_import()
- response = client.get(
+ response = admin_api_client.get(
path=reverse(
"import-artist-detail",
kwargs={"pk": artist_import_job.id},
),
)
+ assert response.status_code == status.HTTP_200_OK, response.data
+ assert response.data["import_finished"]
+
+
+@pytest.mark.django_db(transaction=True)
+def test_force_import_api_detail(
+ admin_api_client: APIClient,
+ superuser: User,
+ force_import_artist_job: ImportJob,
+):
+ """Test detail api for force import job.
+
+ Ensure created import job with invalid file will have `PARSED` status.
+ Ensure force import job after confirmation will skip invalid rows
+ and create right ones.
+
+ """
+ response = admin_api_client.get(
+ path=reverse(
+ "import-artist-detail",
+ kwargs={"pk": force_import_artist_job.id},
+ ),
+ )
+ assert response.status_code == status.HTTP_200_OK, response.data
+ assert response.data["import_status"] == ImportJob.ImportStatus.PARSED
+
+ force_import_artist_job.refresh_from_db()
+ force_import_artist_job.confirm_import()
+
+ response = admin_api_client.get(
+ path=reverse(
+ "import-artist-detail",
+ kwargs={"pk": force_import_artist_job.id},
+ ),
+ )
+ assert response.status_code == status.HTTP_200_OK, response.data
assert response.data["import_finished"]
+
+ force_import_artist_job.refresh_from_db()
+ assert force_import_artist_job.result.totals["new"] == 1
+ assert force_import_artist_job.result.totals["skip"] == 1
diff --git a/tests/test_models/test_import/test_import_data.py b/tests/test_models/test_import/test_import_data.py
index 714c33e..9ff2e1e 100644
--- a/tests/test_models/test_import/test_import_data.py
+++ b/tests/test_models/test_import/test_import_data.py
@@ -82,3 +82,64 @@ def test_import_data_wrong_status(artist_import_job: ImportJob):
match=f"ImportJob with id {artist_import_job.id} has incorrect status",
):
artist_import_job.import_data()
+
+
+@pytest.mark.parametrize(
+ [
+ "skip_parse_step",
+ "is_valid_file",
+ "expected_status",
+ "is_instance_created",
+ ],
+ [
+ [False, True, ImportJob.ImportStatus.PARSED, False],
+ [False, False, ImportJob.ImportStatus.INPUT_ERROR, False],
+ [True, True, ImportJob.ImportStatus.IMPORTED, True],
+ [True, False, ImportJob.ImportStatus.IMPORT_ERROR, False],
+ ],
+)
+@pytest.mark.django_db(transaction=True)
+def test_import_data_skip_parse_step(
+ new_artist: Artist,
+ skip_parse_step: bool,
+ is_valid_file: bool,
+ expected_status: ImportJob.ImportStatus,
+ is_instance_created: bool,
+):
+ """Test import job skip parse step logic.
+
+ If `skip_parse_step=True`,
+ then instance will import data if no errors detected.
+ If `skip_parse_step=False`, then parse only.
+
+ """
+ import_job: ImportJob = ArtistImportJobFactory.build(
+ artists=[new_artist],
+ force_import=False,
+ skip_parse_step=skip_parse_step,
+ is_valid_file=is_valid_file,
+ )
+ import_job.save()
+ import_job.refresh_from_db()
+
+ assert import_job.import_status == expected_status
+ assert (
+ Artist.objects.filter(name=new_artist.name).exists()
+ == is_instance_created
+ )
+
+
+def test_force_import_create_correct_rows(
+ new_artist: Artist,
+):
+ """Test import job with `force_import=True` create correct rows."""
+ import_job: ImportJob = ArtistImportJobFactory(
+ artists=[new_artist],
+ force_import=True,
+ skip_parse_step=True,
+ is_valid_file=False,
+ )
+ import_job.import_data()
+ import_job.refresh_from_db()
+ assert import_job.import_status == import_job.ImportStatus.IMPORTED
+ assert Artist.objects.filter(name=new_artist.name).exists()
diff --git a/tests/test_models/test_import/test_parse_data.py b/tests/test_models/test_import/test_parse_data.py
index 5637de6..e61dbbd 100644
--- a/tests/test_models/test_import/test_parse_data.py
+++ b/tests/test_models/test_import/test_parse_data.py
@@ -3,6 +3,9 @@
from import_export_extensions.models import ImportJob
+from ...fake_app.factories import ArtistImportJobFactory
+from ...fake_app.models import Artist
+
def test_parse_valid_data_file(artist_import_job: ImportJob):
"""Test `parse_data` with valid importing data.
@@ -126,3 +129,37 @@ def test_error_message_length(
artist_import_job.parse_data()
assert len(artist_import_job.error_message) == max_length
+
+
+@pytest.mark.parametrize(
+ [
+ "force_import",
+ "expected_status",
+ ],
+ [
+ [False, ImportJob.ImportStatus.INPUT_ERROR],
+ [True, ImportJob.ImportStatus.PARSED],
+ ],
+)
+@pytest.mark.django_db(transaction=True)
+def test_parse_data_invalid_row_file(
+ new_artist: Artist,
+ force_import: bool,
+ expected_status: ImportJob.ImportStatus,
+):
+ """Test parse file with invalid row.
+
+ If force_import = False, then job must finish with `INPUT_ERROR`.
+ If force_import = True, then job must finish with `PARSED`
+ and skip invalid rows.
+
+ """
+ import_job: ImportJob = ArtistImportJobFactory(
+ artists=[new_artist],
+ force_import=force_import,
+ is_valid_file=False,
+ )
+ import_job.parse_data()
+ import_job.refresh_from_db()
+
+ assert import_job.import_status == expected_status
|