11433529912

Committed 21 Oct 2024 05:02AM UTC coverage: 88.383% (+2.0%) from 86.413%

Build # 11433529912

Build Type

Pull #56

github

Committed by

web-flow

Commit Message

Merge dc0ce208f into f4ea71ece

Pull Request Pull Request #56: Add tests for export job admin progress

Run Details

11 of 13 new or added lines in 2 files covered. (84.62%)

15 existing lines in 1 file now uncovered.

1301 of 1472 relevant lines covered (88.38%)

10.61 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

68.06

/import_export_extensions/admin/model_admins/export_job_admin.py


from http import HTTPStatus

from django.contrib import admin, messages
from django.core.handlers.wsgi import WSGIRequest
from django.db.models import QuerySet
from django.http import JsonResponse
from django.urls import re_path
from django.utils.translation import gettext_lazy as _

from ... import models, tasks
from .. import forms
from . import mixins


class ExportJobAdmin(
    mixins.BaseImportExportJobAdminMixin,
    admin.ModelAdmin,
):
    """Admin class for debugging ExportJob."""

    form = forms.ExportJobAdminForm
    exclude = ("result",)
    list_display = (
        "id",
        "export_status",
        "_model",
        "export_started",
        "export_finished",
        "created",
        "created_by",
    )
    list_display_links = (
        "id",
        "export_status",
        "_model",
    )
    export_job_model = models.ExportJob
    list_filter = ("export_status",)
    list_select_related = ("created_by",)
    actions = (
        "cancel_jobs",
    )

    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.

        /admin/import_export_extensions/exportjob/<job_id>/progress/

        """
        urls = super().get_urls()
        export_urls = [
            re_path(
                route=r"^(?P<job_id>\d+)/progress/$",
                view=self.admin_site.admin_view(self.export_job_progress_view),
                name="export_job_progress",
            ),
        ]
        return export_urls + urls

    def export_job_progress_view(
        self,
        request: WSGIRequest,
        job_id: int,
        **kwargs,
    ):
        """View to return `ExportJob` status as JSON.

        If current status is exporting, view also returns job state
        and percent of completed work.

        Return:
            Response: dictionary with status (optionally, state and percent).

        """
        try:
            job: models.ExportJob = self.export_job_model.objects.get(
                id=job_id,
            )
        except self.export_job_model.DoesNotExist as error:
            return JsonResponse(
                dict(validation_error=error.args[0]),
                status=HTTPStatus.NOT_FOUND,
            )

        response_data = dict(status=job.export_status.title())

        if job.export_status != models.ExportJob.ExportStatus.EXPORTING:
            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(
            dict(
                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.

        """
        readonly_fields = [
            "export_status",
            "traceback",
            "file_format_path",
            "created",
            "export_started",
            "export_finished",
            "error_message",
            "_model",
        ]
        if obj:
            readonly_fields.extend(
                [
                    "resource_path",
                    "data_file",
                    "resource_kwargs",
                ],
            )

        return readonly_fields

    def get_fieldsets(
        self,
        request: WSGIRequest,
        obj: models.ExportJob | None = None,
    ):
        """Get fieldsets depending on object status."""
        status = (
            _("Status"),
            {
                "fields": (
                    "export_status",
                    "_model",
                    "created",
                    "export_started",
                    "export_finished",
                ),
            },
        )
        progress = (
            _("Status"),
            {
                "fields": (
                    "export_status",
                    "export_progressbar",
                ),
            },
        )
        export_params = (
            _("Export params"),
            {
                "fields": (
                    "resource_path",
                    "resource_kwargs",
                    "file_format_path",
                ),
                "classes": ("collapse",),
            },
        )
        traceback_fields = (
            _("Traceback"),
            {
                "fields": (
                    "error_message",
                    "traceback",
                ),
            },
        )
        result = (
            _("Export results"),
            {
                "fields": ("data_file",),
            },
        )

        if (
            not obj
            or obj.export_status == models.ExportJob.ExportStatus.CREATED
        ):
            return [status, export_params]

        if obj.export_status == models.ExportJob.ExportStatus.EXPORTED:
            return [status, result, export_params]

        if obj.export_status == models.ExportJob.ExportStatus.EXPORTING:
            return [progress, export_params]

        return [status, traceback_fields, export_params]

    @admin.action(description="Cancel selected jobs")
    def cancel_jobs(self, request: WSGIRequest, queryset: QuerySet):
        """Admin action for cancelling data export."""
        for job in queryset:
            try:
                job.cancel_export()
                self.message_user(
                    request,
                    _(f"Export of {job} canceled"),
                    messages.SUCCESS,
                )
            except ValueError as error:
                self.message_user(request, str(error), messages.ERROR)


admin.site.register(models.ExportJob, ExportJobAdmin)

1
2	from http import HTTPStatus	12✔
3
4	from django.contrib import admin, messages	12✔
5	from django.core.handlers.wsgi import WSGIRequest	12✔
6	from django.db.models import QuerySet	12✔
7	from django.http import JsonResponse	12✔
8	from django.urls import re_path	12✔
9	from django.utils.translation import gettext_lazy as _	12✔
10
11	from ... import models, tasks	12✔
12	from .. import forms	12✔
13	from . import mixins	12✔
14
15
16	class ExportJobAdmin(	12✔
17	mixins.BaseImportExportJobAdminMixin,
18	admin.ModelAdmin,
19	):
20	"""Admin class for debugging ExportJob."""
21
22	form = forms.ExportJobAdminForm	12✔
23	exclude = ("result",)	12✔
24	list_display = (	12✔
25	"id",
26	"export_status",
27	"_model",
28	"export_started",
29	"export_finished",
30	"created",
31	"created_by",
32	)
33	list_display_links = (	12✔
34	"id",
35	"export_status",
36	"_model",
37	)
38	export_job_model = models.ExportJob	12✔
39	list_filter = ("export_status",)	12✔
40	list_select_related = ("created_by",)	12✔
41	actions = (	12✔
42	"cancel_jobs",
43	)
44
45	def export_data_action(	12✔
46	self,
47	request: WSGIRequest,
48	obj: models.ExportJob,
49	):
50	"""Admin action for starting data export.
51
52	Data export should auto start after export confirmation. But there
53	may be issues with celery and task did not start. So this action
54	for such cases.
55
56	"""
57	tasks.export_data_task.delay(obj.id)	×
58
59	export_data_action.label = _("Start Export")	12✔
60
61	def get_urls(self):	12✔
62	"""Add url to get current export job progress in JSON representation.
63
64	/admin/import_export_extensions/exportjob/<job_id>/progress/
65
66	"""
67	urls = super().get_urls()	12✔
68	export_urls = [	12✔
69	re_path(
70	route=r"^(?P<job_id>\d+)/progress/$",
71	view=self.admin_site.admin_view(self.export_job_progress_view),
72	name="export_job_progress",
73	),
74	]
75	return export_urls + urls	12✔
76
77	def export_job_progress_view(	12✔
78	self,
79	request: WSGIRequest,
80	job_id: int,
81	**kwargs,
82	):
83	"""View to return `ExportJob` status as JSON.
84
85	If current status is exporting, view also returns job state
86	and percent of completed work.
87
88	Return:
89	Response: dictionary with status (optionally, state and percent).
90
91	"""
92	try:	12✔
93	job: models.ExportJob = self.export_job_model.objects.get(	12✔
94	id=job_id,
95	)
96	except self.export_job_model.DoesNotExist as error:	12✔
97	return JsonResponse(	12✔
98	dict(validation_error=error.args[0]),
99	status=HTTPStatus.NOT_FOUND,
100	)
101
102	response_data = dict(status=job.export_status.title())	12✔
103
104	if job.export_status != models.ExportJob.ExportStatus.EXPORTING:	12✔
105	return JsonResponse(response_data)	12✔
106
107	percent = 0	12✔
108	total = 0	12✔
109	current = 0	12✔
110	job_progress = job.progress	12✔
111	progress_info = job_progress["info"]	12✔
112
113	if progress_info and progress_info["total"]:	12✔
114	total = progress_info["total"]	12✔
115	current = progress_info["current"]	12✔
116	percent = int(100 / total * current)	12✔
117
118	response_data.update(	12✔
119	dict(
120	state=job_progress["state"],
121	percent=percent,
122	total=total,
123	current=current,
124	),
125	)
126	return JsonResponse(response_data)	12✔
127
128	def get_readonly_fields(self, request, obj=None):	12✔
129	"""Return readonly fields.
130
131	Some fields are editable for new ExportJob.
132
133	"""
NEW 134	readonly_fields = [	×
135	"export_status",
136	"traceback",
137	"file_format_path",
138	"created",
139	"export_started",
140	"export_finished",
141	"error_message",
142	"_model",
143	]
UNCOV 144	if obj:	×
NEW 145	readonly_fields.extend(	×
146	[
147	"resource_path",
148	"data_file",
149	"resource_kwargs",
150	],
151	)
152
UNCOV 153	return readonly_fields	×
154
155	def get_fieldsets(	12✔
156	self,
157	request: WSGIRequest,
158	obj: models.ExportJob \| None = None,
159	):
160	"""Get fieldsets depending on object status."""
UNCOV 161	status = (	×
162	_("Status"),
163	{
164	"fields": (
165	"export_status",
166	"_model",
167	"created",
168	"export_started",
169	"export_finished",
170	),
171	},
172	)
UNCOV 173	progress = (	×
174	_("Status"),
175	{
176	"fields": (
177	"export_status",
178	"export_progressbar",
179	),
180	},
181	)
UNCOV 182	export_params = (	×
183	_("Export params"),
184	{
185	"fields": (
186	"resource_path",
187	"resource_kwargs",
188	"file_format_path",
189	),
190	"classes": ("collapse",),
191	},
192	)
UNCOV 193	traceback_fields = (	×
194	_("Traceback"),
195	{
196	"fields": (
197	"error_message",
198	"traceback",
199	),
200	},
201	)
UNCOV 202	result = (	×
203	_("Export results"),
204	{
205	"fields": ("data_file",),
206	},
207	)
208
UNCOV 209	if (	×
210	not obj
211	or obj.export_status == models.ExportJob.ExportStatus.CREATED
212	):
213	return [status, export_params]	×
214
215	if obj.export_status == models.ExportJob.ExportStatus.EXPORTED:	×
216	return [status, result, export_params]	×
217
218	if obj.export_status == models.ExportJob.ExportStatus.EXPORTING:	×
UNCOV 219	return [progress, export_params]	×
220
UNCOV 221	return [status, traceback_fields, export_params]	×
222
223	@admin.action(description="Cancel selected jobs")	12✔
224	def cancel_jobs(self, request: WSGIRequest, queryset: QuerySet):	12✔
225	"""Admin action for cancelling data export."""
226	for job in queryset:	×
UNCOV 227	try:	×
UNCOV 228	job.cancel_export()	×
UNCOV 229	self.message_user(	×
230	request,
231	_(f"Export of {job} canceled"),
232	messages.SUCCESS,
233	)
UNCOV 234	except ValueError as error:	×
UNCOV 235	self.message_user(request, str(error), messages.ERROR)	×
236
237
238	admin.site.register(models.ExportJob, ExportJobAdmin)	12✔

saritasa-nest / django-import-export-extensions / 11433529912

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous