Merge pull request #58 from saritasa-nest/feature/improve-coverage

Add tests for import admin actions

import_export_extensions/admin/model_admins/export_job_admin.py

@@ -41,6 +41,19 @@ class ExportJobAdmin(
     actions = (
         "cancel_jobs",
     )
+    readonly_fields = (
+        "export_status",
+        "traceback",
+        "file_format_path",
+        "created",
+        "export_started",
+        "export_finished",
+        "error_message",
+        "_model",
+        "resource_path",
+        "data_file",
+        "resource_kwargs",
+    )
 
     def export_data_action(
         self,
@@ -116,43 +129,17 @@ def export_job_progress_view(
             percent = int(100 / total * current)
 
         response_data.update(
-            dict(
-                state=job_progress["state"],
-                percent=percent,
-                total=total,
-                current=current,
-            ),
+            state=job_progress["state"],
+            percent=percent,
+            total=total,
+            current=current,
         )
         return JsonResponse(response_data)
 
-    def get_readonly_fields(self, request, obj=None):
-        """Return readonly fields.
-
-        Some fields are editable for new ExportJob.
-
-        """
-        base_readonly_fields = super().get_readonly_fields(request, obj)
-        readonly_fields = (
-            *base_readonly_fields,
-            "export_status",
-            "traceback",
-            "file_format_path",
-            "created",
-            "export_started",
-            "export_finished",
-            "error_message",
-            "_model",
-            "resource_path",
-            "data_file",
-            "resource_kwargs",
-        )
-
-        return readonly_fields
-
     def get_fieldsets(
         self,
         request: WSGIRequest,
-        obj: models.ExportJob | None = None,
+        obj: models.ExportJob,
     ):
         """Get fieldsets depending on object status."""
         status = (
@@ -203,10 +190,7 @@ def get_fieldsets(
             },
         )
 
-        if (
-            not obj
-            or obj.export_status == models.ExportJob.ExportStatus.CREATED
-        ):
+        if obj.export_status == models.ExportJob.ExportStatus.CREATED:
             return [status, export_params]
 
         if obj.export_status == models.ExportJob.ExportStatus.EXPORTED:

import_export_extensions/admin/model_admins/import_job_admin.py

@@ -1,3 +1,4 @@
+import http
 
 from django.contrib import admin, messages
 from django.core.handlers.wsgi import WSGIRequest
@@ -42,6 +43,22 @@ class ImportJobAdmin(
         "cancel_jobs",
         "confirm_jobs",
     )
+    readonly_fields = (
+        "import_status",
+        "_model",
+        "created_by",
+        "traceback",
+        "_show_results",
+        "_input_errors",
+        "created",
+        "parse_finished",
+        "import_started",
+        "import_finished",
+        "resource_path",
+        "input_errors_file",
+        "data_file",
+        "resource_kwargs",
+    )
 
     def get_queryset(self, request: WSGIRequest):
         """Override `get_queryset`.
@@ -88,66 +105,36 @@ def import_job_progress_view(
                 id=job_id,
             )
         except self.import_job_model.DoesNotExist as error:
-            return JsonResponse(dict(validation_error=error.args[0]))
+            return JsonResponse(
+                dict(validation_error=error.args[0]),
+                status=http.HTTPStatus.NOT_FOUND,
+            )
 
         response_data = dict(status=job.import_status.title())
 
-        if job.import_status in models.ImportJob.progress_statuses:
-            percent = 0
-            total = 0
-            current = 0
-            info = job.progress["info"]
-
-            if info and info["total"]:
-                percent = int(100 / info["total"] * info["current"])
-                total = info["total"]
-                current = info["current"]
-
-            response_data.update(
-                dict(
-                    state=job.progress["state"],
-                    percent=percent,
-                    total=total,
-                    current=current,
-                ),
-            )
+        if job.import_status not in models.ImportJob.progress_statuses:
+            return JsonResponse(response_data)
+
+        percent = 0
+        total = 0
+        current = 0
+        job_progress = job.progress
+        progress_info = job_progress["info"]
+
+        if progress_info and progress_info["total"]:
+            total = progress_info["total"]
+            current = progress_info["current"]
+            percent = int(100 / total * current)
+
+        response_data.update(
+            state=job_progress["state"],
+            percent=percent,
+            total=total,
+            current=current,
+        )
 
         return JsonResponse(response_data)
 
-    def get_readonly_fields(
-        self,
-        request: WSGIRequest,
-        obj: models.ImportJob | None = None,
-    ) -> list[str]:
-        """Return readonly fields.
-
-        Some fields are editable for new ImportJob.
-
-        """
-        readonly_fields = [
-            "import_status",
-            "_model",
-            "created_by",
-            "traceback_str",
-            "_show_results",
-            "_input_errors",
-            "created",
-            "parse_finished",
-            "import_started",
-            "import_finished",
-        ]
-        if obj:
-            readonly_fields.extend(
-                [
-                    "resource_path",
-                    "input_errors_file",
-                    "data_file",
-                    "resource_kwargs",
-                ],
-            )
-
-        return readonly_fields
-
     def _show_results(
         self,
         obj: models.ImportJob | None = None,
@@ -188,7 +175,7 @@ def _input_errors(self, job: models.ImportJob):
     def get_fieldsets(
         self,
         request: WSGIRequest,
-        obj: models.ImportJob | None = None,
+        obj: models.ImportJob,
     ):
         """Get fieldsets depending on object status."""
         status = (
@@ -249,13 +236,10 @@ def get_fieldsets(
             },
         )
 
-        if not obj:
-            return [status, import_params]
-
         if obj.import_status == models.ImportJob.ImportStatus.CREATED:
             return [status, import_params]
 
-        if obj.import_status in models.ImportJob.results_statuses:
+        if obj.import_status in models.ImportJob.success_statuses:
             return [status, result, data, import_params]
 
         if obj.import_status in models.ImportJob.progress_statuses:

import_export_extensions/models/import_job.py

@@ -249,47 +249,20 @@ def resource(self) -> CeleryResource:
 
     @property
     def progress(self) -> TaskStateInfo | None:
-        """Return dict with parsing state.
-
-        Example for sync mode::
-
-            {
-                'state': 'PARSING',
-                'info': None
-            }
-
-        Example for background (celery) mode::
-
-            {
-                'state': 'PARSING',
-                'info': {'current': 15, 'total': 100}
-            }
-
-        Possible states:
-            1. PENDING
-            2. STARTED
-            3. SUCCESS
-            4. PARSING - custom status that also set importing info
-
-        https://docs.celeryproject.org/en/latest/userguide/tasks.html#states
-
-        """
-        if self.import_status not in self.progress_statuses:
-            return None
-
-        current_task = (
+        """Return dict with parsing state."""
+        current_task_id = (
             self.parse_task_id
             if self.import_status == self.ImportStatus.PARSING
             else self.import_task_id
         )
 
-        if not current_task or current_app.conf.task_always_eager:
-            return dict(
-                state=self.import_status.upper(),
-                info=None,
-            )
+        if (
+            not current_task_id
+            or self.import_status not in self.progress_statuses
+        ):
+            return None
 
-        return self._get_task_state(current_task)
+        return self._get_task_state(current_task_id)
 
     def _check_import_status_correctness(
         self,

import_export_extensions/widgets.py

@@ -86,7 +86,7 @@ def render(
     ) -> str:
         """Return an export representation of a intermediate instances.
 
-        For atrists example should be returned something like
+        For artists example should be returned something like
             "5:1990-12-12;19:2005-08-16"
             where 5 is band id
 

test_project/tests/integration_tests/test_admin/test_export.py

@@ -126,11 +126,7 @@ def test_export_progress_with_deleted_export_job(
     superuser: User,
     mocker: pytest_mock.MockerFixture,
 ):
-    """Test export job admin progress page with deleted export job.
-
-    Check that page available, but return an error message.
-
-    """
+    """Test export job admin progress page with deleted export job."""
     client.force_login(superuser)
 
     mocker.patch("import_export_extensions.tasks.export_data_task.apply_async")
@@ -213,13 +209,14 @@ def test_cancel_export_admin_action(
             "action": "cancel_jobs",
             "_selected_action": [job.pk],
         },
+        follow=True,
     )
     job.refresh_from_db()
 
-    assert response.status_code == status.HTTP_302_FOUND
+    assert response.status_code == status.HTTP_200_OK
     assert job.export_status == ExportJob.ExportStatus.CANCELLED
     assert (
-        response.wsgi_request._messages._queued_messages[0].message
+        response.context["messages"]._loaded_data[0].message
         == f"Export of {job} canceled"
     )
     export_data_mock.assert_called_once()
@@ -246,14 +243,15 @@ def test_cancel_export_admin_action_with_incorrect_export_job_status(
             "action": "cancel_jobs",
             "_selected_action": [job.pk],
         },
+        follow=True,
     )
     job.refresh_from_db()
 
-    assert response.status_code == status.HTTP_302_FOUND
+    assert response.status_code == status.HTTP_200_OK
     assert job.export_status == ExportJob.ExportStatus.EXPORTED
     assert (
         expected_error_message
-        in response.wsgi_request._messages._queued_messages[0].message
+        in response.context["messages"]._loaded_data[0].message
     )
     revoke_mock.assert_not_called()