Prepare tests for celery admin mixins

import_export_extensions/admin/mixins/base_mixin.py

@@ -0,0 +1,22 @@
+import typing
+
+from import_export import admin as import_export_admin
+
+from . import types
+
+
+class BaseCeleryImportExportAdminMixin(
+    import_export_admin.ImportExportMixinBase,
+):
+    """Extend base mixin with common logic for import/export."""
+
+    @property
+    def model_info(self) -> types.ModelInfo:
+        """Get info of model."""
+        return types.ModelInfo(
+            meta=self.model._meta,
+        )
+
+    def get_context_data(self, **kwargs) -> dict[str, typing.Any]:
+        """Get context data."""
+        return {}

import_export_extensions/admin/mixins/export_mixin.py

@@ -4,25 +4,24 @@
 from django.core.handlers.wsgi import WSGIRequest
 from django.http import (
     HttpResponse,
-    HttpResponseForbidden,
     HttpResponseRedirect,
 )
 from django.shortcuts import get_object_or_404
 from django.template.response import TemplateResponse
 from django.urls import re_path, reverse
 from django.utils.translation import gettext_lazy as _
 
-from import_export import admin as base_admin
-from import_export import forms as base_forms
-from import_export import mixins as base_mixins
+from import_export import admin as import_export_admin
+from import_export import forms as import_export_forms
+from import_export import mixins as import_export_mixins
 
 from ... import models
-from . import types
+from . import base_mixin, types
 
 
 class CeleryExportAdminMixin(
-    base_mixins.BaseExportMixin,
-    base_admin.ImportExportMixinBase,
+    import_export_mixins.BaseExportMixin,
+    base_mixin.BaseCeleryImportExportAdminMixin,
 ):
     """Admin mixin for celery export.
 
@@ -63,22 +62,13 @@ class CeleryExportAdminMixin(
     export_results_statuses = models.ExportJob.export_finished_statuses
 
     # Copy methods of mixin from original package to reuse it here
-    has_export_permission = base_admin.ExportMixin.has_export_permission
+    has_export_permission = (
+        import_export_admin.ExportMixin.has_export_permission
+    )
 
-    @property
-    def model_info(self) -> types.ModelInfo:
-        """Get info of exported model."""
-        return types.ModelInfo(
-            meta=self.model._meta,
-        )
-
-    def get_context_data(
-        self,
-        request: WSGIRequest,
-        **kwargs,
-    ) -> dict[str, typing.Any]:
-        """Get context data."""
-        return {}
+    def get_export_context_data(self, **kwargs):
+        """Get context data for export."""
+        return self.get_context_data(**kwargs)
 
     def get_urls(self):
         """Return list of urls.
@@ -130,7 +120,7 @@ def celery_export_action(self, request, *args, **kwargs):
             raise PermissionDenied
 
         formats = self.get_export_formats()
-        form = base_forms.ExportForm(
+        form = import_export_forms.ExportForm(
             formats=formats,
             resources=self.get_export_resource_classes(request),
             data=request.POST or None,
@@ -158,7 +148,7 @@ def celery_export_action(self, request, *args, **kwargs):
             )
 
         # GET: display Export Form
-        context = self.get_context_data(request)
+        context = self.get_export_context_data()
         context.update(self.admin_site.each_context(request))
 
         context["title"] = _("Export")
@@ -196,7 +186,7 @@ def export_job_status_view(
                 job=job,
             )
 
-        context = self.get_context_data(request)
+        context = self.get_export_context_data()
         job_url = reverse("admin:export_job_progress", args=(job.id,))
 
         context["title"] = _("Export status")
@@ -235,10 +225,7 @@ def export_job_results_view(
                 job=job,
             )
 
-        context = self.get_context_data(request)
-
-        if request.method != "GET":
-            return HttpResponseForbidden()
+        context = self.get_export_context_data()
 
         # GET request, show export results
         context["title"] = _("Export results")
@@ -293,8 +280,7 @@ def _redirect_to_export_status_page(
         )
         url = reverse(url_name, kwargs=dict(job_id=job.id))
         query = request.GET.urlencode()
-        if query:
-            url = f"{url}?{query}"
+        url = f"{url}?{query}" if query else url
         return HttpResponseRedirect(redirect_to=url)
 
     def _redirect_to_export_results_page(
@@ -308,8 +294,7 @@ def _redirect_to_export_results_page(
         )
         url = reverse(url_name, kwargs=dict(job_id=job.id))
         query = request.GET.urlencode()
-        if query:
-            url = f"{url}?{query}"
+        url = f"{url}?{query}" if query else url
         return HttpResponseRedirect(redirect_to=url)
 
     def changelist_view(
@@ -319,5 +304,5 @@ def changelist_view(
     ):
         """Add the check for permission to changelist template context."""
         context = context or {}
-        context["has_export_permission"] = True
+        context["has_export_permission"] = self.has_export_permission(request)
         return super().changelist_view(request, context)

import_export_extensions/admin/mixins/import_mixin.py

@@ -1,7 +1,6 @@
 import typing
 
 from django.conf import settings
-from django.core.cache import cache
 from django.core.exceptions import PermissionDenied
 from django.core.handlers.wsgi import WSGIRequest
 from django.forms.forms import Form
@@ -15,17 +14,17 @@
 from django.urls import re_path, reverse
 from django.utils.translation import gettext_lazy as _
 
-from import_export import admin as base_admin
-from import_export import mixins as base_mixins
+from import_export import admin as import_export_admin
+from import_export import mixins as import_export_mixins
 
 from ... import models
 from ..forms import ForceImportForm
-from . import types
+from . import base_mixin, types
 
 
 class CeleryImportAdminMixin(
-    base_mixins.BaseImportMixin,
-    base_admin.ImportExportMixinBase,
+    import_export_mixins.BaseImportMixin,
+    base_mixin.BaseCeleryImportExportAdminMixin,
 ):
     """Admin mixin for celery import.
 
@@ -75,30 +74,17 @@ class CeleryImportAdminMixin(
 
     skip_admin_log = None
     # Copy methods of mixin from original package to reuse it here
-    generate_log_entries = base_admin.ImportMixin.generate_log_entries
-    get_skip_admin_log = base_admin.ImportMixin.get_skip_admin_log
-    has_import_permission = base_admin.ImportMixin.has_import_permission
-    _log_actions = base_admin.ImportMixin._log_actions
-    _create_log_entries = base_admin.ImportMixin._create_log_entries
-    _create_log_entry = base_admin.ImportMixin._create_log_entry
-
-    @property
-    def model_info(self) -> types.ModelInfo:
-        """Get info of imported model."""
-        return types.ModelInfo(
-            meta=self.model._meta,
-        )
-
-    def get_context_data(
-        self,
-        request: WSGIRequest,
-        **kwargs,
-    ) -> dict[str, typing.Any]:
-        """Get context data."""
-        return {}
+    generate_log_entries = import_export_admin.ImportMixin.generate_log_entries
+    get_skip_admin_log = import_export_admin.ImportMixin.get_skip_admin_log
+    has_import_permission = (
+        import_export_admin.ImportMixin.has_import_permission
+    )
+    _log_actions = import_export_admin.ImportMixin._log_actions
+    _create_log_entries = import_export_admin.ImportMixin._create_log_entries
+    _create_log_entry = import_export_admin.ImportMixin._create_log_entry
 
     def get_import_context_data(self, **kwargs):
-        """Get context for import data."""
+        """Get context data for import."""
         return self.get_context_data(**kwargs)
 
     def get_urls(self):
@@ -159,7 +145,7 @@ def celery_import_action(
         if not self.has_import_permission(request):
             raise PermissionDenied
 
-        context = self.get_context_data(request)
+        context = self.get_import_context_data()
         resource_classes = self.get_import_resource_classes(request)
 
         form = ForceImportForm(
@@ -238,7 +224,7 @@ def celery_import_job_status_view(
                 job=job,
             )
 
-        context = self.get_context_data(request)
+        context = self.get_import_context_data()
         job_url = reverse("admin:import_job_progress", args=(job.id,))
         context.update(
             dict(
@@ -283,7 +269,7 @@ def celery_import_job_results_view(
                 job=job,
             )
 
-        context = self.get_context_data(request=request)
+        context = self.get_import_context_data()
 
         if request.method == "GET":
             # GET request, show parse results
@@ -292,14 +278,14 @@ def celery_import_job_results_view(
             context["result"] = result
             context["title"] = _("Import results")
 
-            if job.import_status != models.ImportJob.ImportStatus.PARSED:
+            if job.import_status == models.ImportJob.ImportStatus.PARSED:
+                context["confirm_form"] = Form()
+            else:
                 # display import form
                 context["import_form"] = ForceImportForm(
                     formats=self.get_import_formats(),
                     resources=self.get_import_resource_classes(request),
                 )
-            else:
-                context["confirm_form"] = Form()
 
             context.update(self.admin_site.each_context(request))
             context["opts"] = self.model_info.meta
@@ -310,17 +296,21 @@ def celery_import_job_results_view(
                 context,
             )
 
-        # POST request. If data is invalid - error
-        if job.import_status != models.ImportJob.ImportStatus.PARSED:
-            return HttpResponseForbidden(
-                "Data invalid, before importing data "
-                "needs to be successfully parsed."
-                f"Current status: {job.import_status}",
+        # POST request
+        if job.import_status == models.ImportJob.ImportStatus.PARSED:
+            # start celery task for data importing
+            job.confirm_import()
+            return self._redirect_to_import_status_page(
+                request=request,
+                job=job,
             )
 
-        # start celery task for data importing
-        job.confirm_import()
-        return self._redirect_to_import_status_page(request=request, job=job)
+        return HttpResponseForbidden(
+            "Data invalid, before importing data "
+            "needs to be successfully parsed. "
+            f"Current status: {job.import_status}",
+        )
+
 
     def create_import_job(
         self,
@@ -384,20 +374,6 @@ def _redirect_to_results_page(
         if job.import_status != models.ImportJob.ImportStatus.PARSED:
             return HttpResponseRedirect(redirect_to=url)
 
-        # Redirections add one by one links to `redirect_to`
-        key = request.session.get("redirect_key", None)
-        session = request.session
-        if key:
-            links = cache.get(key)
-            try:
-                session["redirect_to"] = links[0]
-                del links[0]
-                cache.set(key, links)
-            except (TypeError, IndexError):
-                session.pop("redirect_to", None)
-                session.pop("redirect_key", None)
-                cache.delete(key)
-
         return HttpResponseRedirect(redirect_to=url)
 
     def changelist_view(

import_export_extensions/admin/model_admins/export_job_admin.py

@@ -8,7 +8,7 @@
 from django.urls import re_path
 from django.utils.translation import gettext_lazy as _
 
-from ... import models, tasks
+from ... import models
 from .. import forms
 from . import mixins
 
@@ -55,22 +55,6 @@ class ExportJobAdmin(
         "resource_kwargs",
     )
 
-    def export_data_action(
-        self,
-        request: WSGIRequest,
-        obj: models.ExportJob,
-    ):
-        """Admin action for starting data export.
-
-        Data export should auto start after export confirmation. But there
-        may be issues with celery and task did not start. So this action
-        for such cases.
-
-        """
-        tasks.export_data_task.delay(obj.id)
-
-    export_data_action.label = _("Start Export")
-
     def get_urls(self):
         """Add url to get current export job progress in JSON representation.
 

import_export_extensions/admin/model_admins/import_job_admin.py

@@ -137,7 +137,7 @@ def import_job_progress_view(
 
     def _show_results(
         self,
-        obj: models.ImportJob | None = None,
+        obj: models.ImportJob,
     ) -> str:
         """Show results totals.
 
@@ -149,9 +149,6 @@ def _show_results(
         Error: 0
 
         """
-        if not obj:
-            return ""
-
         result_sections = []
         for key, value in obj.result.totals.items():
             status_template = f"{key.title()}: {value}"