11511363580

Committed 25 Oct 2024 03:30AM UTC coverage: 95.711% (+5.0%) from 90.687%

Build # 11511363580

Build Type

Pull #58

github

Committed by

web-flow

Commit Message

Merge 77a3934c8 into 6e02e184c

Pull Request Pull Request #58: Add tests for import admin actions

Run Details

16 of 16 new or added lines in 3 files covered. (100.0%)

2 existing lines in 2 files now uncovered.

1406 of 1469 relevant lines covered (95.71%)

11.49 hits per line

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

98.59

/import_export_extensions/admin/model_admins/export_job_admin.py


import http

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=http.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.

        """
        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,
    ):
        """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	import http	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	"""
UNCOV 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=http.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	"""
134	base_readonly_fields = super().get_readonly_fields(request, obj)	12✔
135	readonly_fields = (	12✔
136	*base_readonly_fields,
137	"export_status",
138	"traceback",
139	"file_format_path",
140	"created",
141	"export_started",
142	"export_finished",
143	"error_message",
144	"_model",
145	"resource_path",
146	"data_file",
147	"resource_kwargs",
148	)
149
150	return readonly_fields	12✔
151
152	def get_fieldsets(	12✔
153	self,
154	request: WSGIRequest,
155	obj: models.ExportJob \| None = None,
156	):
157	"""Get fieldsets depending on object status."""
158	status = (	12✔
159	_("Status"),
160	{
161	"fields": (
162	"export_status",
163	"_model",
164	"created",
165	"export_started",
166	"export_finished",
167	),
168	},
169	)
170	progress = (	12✔
171	_("Status"),
172	{
173	"fields": (
174	"export_status",
175	"export_progressbar",
176	),
177	},
178	)
179	export_params = (	12✔
180	_("Export params"),
181	{
182	"fields": (
183	"resource_path",
184	"resource_kwargs",
185	"file_format_path",
186	),
187	"classes": ("collapse",),
188	},
189	)
190	traceback_fields = (	12✔
191	_("Traceback"),
192	{
193	"fields": (
194	"error_message",
195	"traceback",
196	),
197	},
198	)
199	result = (	12✔
200	_("Export results"),
201	{
202	"fields": ("data_file",),
203	},
204	)
205
206	if (	12✔
207	not obj
208	or obj.export_status == models.ExportJob.ExportStatus.CREATED
209	):
210	return [status, export_params]	12✔
211
212	if obj.export_status == models.ExportJob.ExportStatus.EXPORTED:	12✔
213	return [status, result, export_params]	12✔
214
215	if obj.export_status == models.ExportJob.ExportStatus.EXPORTING:	12✔
216	return [progress, export_params]	12✔
217
218	return [status, traceback_fields, export_params]	12✔
219
220	@admin.action(description="Cancel selected jobs")	12✔
221	def cancel_jobs(self, request: WSGIRequest, queryset: QuerySet):	12✔
222	"""Admin action for cancelling data export."""
223	for job in queryset:	12✔
224	try:	12✔
225	job.cancel_export()	12✔
226	self.message_user(	12✔
227	request,
228	_(f"Export of {job} canceled"),
229	messages.SUCCESS,
230	)
231	except ValueError as error:	12✔
232	self.message_user(request, str(error), messages.ERROR)	12✔
233
234
235	admin.site.register(models.ExportJob, ExportJobAdmin)	12✔

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

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