12313892613

Committed 13 Dec 2024 10:24AM UTC coverage: 83.611% (-1.3%) from 84.861%

Build # 12313892613

Build Type

Pull #1348

github

Committed by

web-flow

Commit Message

Merge 7ba1fbea3 into 481a19acc

Pull Request Pull Request #1348: add vue recurrence picker

Run Details

2972 of 3523 branches covered (84.36%)

Branch coverage included in aggregate %.

83 of 358 new or added lines in 9 files covered. (23.18%)

6 existing lines in 1 file now uncovered.

12078 of 14477 relevant lines covered (83.43%)

0.83 hits per line

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

69.3

/ephios/core/views/shift.py

from copy import copy
from datetime import datetime, timedelta

from csp.decorators import csp_exempt
from django.contrib import messages
from django.core.exceptions import ValidationError
from django.shortcuts import get_object_or_404, redirect, render
from django.urls import reverse
from django.utils import timezone
from django.utils.datastructures import MultiValueDictKeyError
from django.utils.timezone import get_default_timezone
from django.utils.translation import gettext as _
from django.views import View
from django.views.generic import DeleteView, FormView, TemplateView
from django.views.generic.detail import SingleObjectMixin

from ephios.core.forms.events import ShiftCopyForm, ShiftForm
from ephios.core.models import Event, Shift
from ephios.core.signals import shift_forms
from ephios.core.signup.flow import enabled_signup_flows, signup_flow_from_slug
from ephios.core.signup.structure import enabled_shift_structures, shift_structure_from_slug
from ephios.extra.mixins import CustomPermissionRequiredMixin, PluginFormMixin


class ShiftCreateView(CustomPermissionRequiredMixin, PluginFormMixin, TemplateView):
    permission_required = "core.change_event"
    template_name = "core/shift_form.html"

    def setup(self, request, *args, **kwargs):
        super().setup(request, *args, **kwargs)
        self.event = get_object_or_404(Event.all_objects, pk=self.kwargs.get("pk"))

    def get_permission_object(self):
        return self.event

    def get_shift_form(self):
        return ShiftForm(self.request.POST or None)

    def get_context_data(self, **kwargs):
        kwargs.setdefault("event", self.event)
        kwargs.setdefault("form", self.get_shift_form())
        kwargs.setdefault("flow_configuration_form", "")
        kwargs.setdefault("structure_configuration_form", "")
        return super().get_context_data(**kwargs)

    def get_plugin_forms(self):
        return shift_forms.send(
            sender=None, shift=getattr(self, "object", None), request=self.request
        )

    def post(self, *args, **kwargs):
        form = self.get_shift_form()
        self.object = form.instance

        try:
            signup_flow = signup_flow_from_slug(
                self.request.POST.get("signup_flow_slug"), event=self.event
            )
            structure = shift_structure_from_slug(
                self.request.POST.get("structure_slug"),
                event=self.event,
            )
        except KeyError:
            if not list(enabled_signup_flows()):
                form.add_error(
                    "signup_flow_slug",
                    _("You must enable plugins providing signup flows to continue."),
                )
            if not list(enabled_shift_structures()):
                form.add_error(
                    "structure_slug",
                    _("You must enable plugins providing shift structures to continue."),
                )
            return self.render_to_response(
                self.get_context_data(
                    form=form,
                )
            )
        except ValueError as e:
            form.add_error(None, ValidationError(e))
            return self.render_to_response(
                self.get_context_data(
                    form=form,
                )
            )

        flow_configuration_form = signup_flow.get_configuration_form(
            self.request.POST, event=self.event, request=self.request
        )
        structure_configuration_form = structure.get_configuration_form(
            self.request.POST, event=self.event, request=self.request
        )
        if not all(
            [
                self.is_valid(form),
                flow_configuration_form.is_valid(),
                structure_configuration_form.is_valid(),
            ]
        ):
            return self.render_to_response(
                self.get_context_data(
                    form=form,
                    flow_configuration_form=flow_configuration_form,
                    structure_configuration_form=structure_configuration_form,
                )
            )

        shift = form.save(commit=False)
        shift.event = self.event
        shift.signup_flow_configuration = flow_configuration_form.cleaned_data
        shift.structure_configuration = structure_configuration_form.cleaned_data
        shift.save()
        self.save_plugin_forms()
        if "addAnother" in self.request.POST:
            return redirect(reverse("core:event_createshift", kwargs={"pk": self.kwargs.get("pk")}))
        if "copyShift" in self.request.POST:
            return redirect(reverse("core:shift_copy", kwargs={"pk": shift.pk}))
        try:
            self.event.activate()
            messages.success(
                self.request,
                _("The event {title} has been saved.").format(title=self.event.title),
            )
        except ValidationError as e:
            messages.error(self.request, e)
        return redirect(self.event.get_absolute_url())


class AbstractShiftConfigurationFormView(CustomPermissionRequiredMixin, SingleObjectMixin, View):
    queryset = Event.all_objects
    permission_required = "core.change_event"
    pk_url_kwarg = "event_id"

    def get(self, request, *args, **kwargs):
        try:
            shift = self.get_object().shifts.get(pk=request.GET.get("shift_id") or None)
        except Shift.DoesNotExist:
            shift = None
        form = self.get_form(self.request, self.get_object(), shift=shift)
        return render(
            request,
            "core/fragments/shift_signup_config_form.html",
            {
                "form": form,
            },
        )

    def get_form(self, request, event, shift=None):
        raise NotImplementedError


class SignupFlowConfigurationFormView(AbstractShiftConfigurationFormView):
    def get_form(
        self,
        request,
        event,
        shift=None,
    ):
        return signup_flow_from_slug(
            self.kwargs.get("slug"), event=event, shift=shift
        ).get_configuration_form(request=request)


class ShiftStructureConfigurationFormView(AbstractShiftConfigurationFormView):
    def get_form(self, request, event, shift=None):
        return shift_structure_from_slug(
            self.kwargs.get("slug"), event=event, shift=shift
        ).get_configuration_form(request=request)


class ShiftUpdateView(
    CustomPermissionRequiredMixin, PluginFormMixin, SingleObjectMixin, TemplateView
):
    model = Shift
    template_name = "core/shift_form.html"
    permission_required = "core.change_event"

    def get_permission_object(self):
        return self.get_object().event

    def get_shift_form(self):
        return ShiftForm(
            self.request.POST or None,
            instance=self.object,
            initial={
                "date": self.object.meeting_time.date(),
                "meeting_time": self.object.meeting_time.astimezone(get_default_timezone()).time(),
                "start_time": self.object.start_time.astimezone(get_default_timezone()).time(),
                "end_time": self.object.end_time.astimezone(get_default_timezone()).time(),
            },
        )

    def get_flow_configuration_form(self):
        return self.object.signup_flow.get_configuration_form(
            request=self.request, data=self.request.POST or None
        )

    def get_structure_configuration_form(self):
        return self.object.structure.get_configuration_form(
            request=self.request, data=self.request.POST or None
        )

    def get_context_data(self, **kwargs):
        self.object = self.get_object()
        kwargs.setdefault("event", self.object.event)
        kwargs.setdefault("form", self.get_shift_form())
        kwargs.setdefault("flow_configuration_form", self.get_flow_configuration_form())
        kwargs.setdefault("structure_configuration_form", self.get_structure_configuration_form())
        return super().get_context_data(**kwargs)

    def get_plugin_forms(self):
        return shift_forms.send(sender=None, shift=self.object, request=self.request)

    def post(self, *args, **kwargs):
        self.object = self.get_object()
        form = self.get_shift_form()
        flow_configuration_form = self.get_flow_configuration_form()
        structure_configuration_form = self.get_structure_configuration_form()
        try:
            signup_flow = signup_flow_from_slug(
                self.request.POST["signup_flow_slug"], shift=self.object
            )
            flow_configuration_form = signup_flow.get_configuration_form(
                self.request.POST,
                event=self.object.event,
                request=self.request,
            )
            structure = shift_structure_from_slug(
                self.request.POST["structure_slug"], shift=self.object
            )
            structure_configuration_form = structure.get_configuration_form(
                self.request.POST,
                event=self.object.event,
                request=self.request,
            )
        except (ValueError, MultiValueDictKeyError):
            pass
        else:
            if all(
                [
                    self.is_valid(form),
                    flow_configuration_form.is_valid(),
                    structure_configuration_form.is_valid(),
                ]
            ):
                shift = form.save(commit=False)
                shift.signup_flow_configuration = flow_configuration_form.cleaned_data
                shift.structure_configuration = structure_configuration_form.cleaned_data
                shift.save()
                self.save_plugin_forms()
                if "addAnother" in self.request.POST:
                    return redirect(
                        reverse("core:event_createshift", kwargs={"pk": shift.event.pk})
                    )
                if "copyShift" in self.request.POST:
                    return redirect(reverse("core:shift_copy", kwargs={"pk": shift.pk}))
                messages.success(
                    self.request, _("The shift {shift} has been saved.").format(shift=shift)
                )
                return redirect(self.object.event.get_absolute_url())

        return self.render_to_response(
            self.get_context_data(
                form=form,
                flow_configuration_form=flow_configuration_form,
                structure_configuration_form=structure_configuration_form,
            )
        )


class ShiftDeleteView(CustomPermissionRequiredMixin, DeleteView):
    permission_required = "core.change_event"
    model = Shift

    def setup(self, request, *args, **kwargs):
        super().setup(request, *args, **kwargs)
        self.object = self.get_object()

    def form_valid(self, form):
        if self.object.event.shifts.count() == 1:
            messages.error(self.request, _("You cannot delete the last shift!"))
            return redirect(self.object.event.get_absolute_url())
        return super().form_valid(form)

    def get_success_url(self):
        messages.success(self.request, _("The shift has been deleted."))
        return self.object.event.get_absolute_url()

    def get_permission_object(self):
        return self.object.event


class ShiftCopyView(CustomPermissionRequiredMixin, SingleObjectMixin, FormView):
    permission_required = "core.change_event"
    model = Shift
    template_name = "core/shift_copy.html"
    form_class = ShiftCopyForm

    def setup(self, request, *args, **kwargs):
        super().setup(request, *args, **kwargs)
        self.object = self.get_object()

    def get_form_kwargs(self):
        kwargs = super().get_form_kwargs()
        kwargs["original_start"] = timezone.localtime(self.object.start_time).isoformat()
        return kwargs

    def get_success_url(self):
        return reverse("core:event_detail", kwargs={"pk": self.object.event.pk, "slug": "event"})

    def form_valid(self, form):
        shift = self.object
        duration = shift.end_time - shift.start_time
        meeting_offset = shift.start_time - shift.meeting_time
        shifts_to_create = []
        recurr = form.cleaned_data["recurrence"]
        tz = timezone.get_current_timezone()
        for dt in recurr.xafter(datetime.now() - timedelta(days=365 * 100), 1000, inc=True):
            dt = timezone.make_aware(dt, tz)
            shift = copy(shift)
            shift.pk = None
            shift.meeting_time = dt - meeting_offset
            shift.start_time = dt
            shift.end_time = dt + duration
            shifts_to_create.append(shift)
        Shift.objects.bulk_create(shifts_to_create)
        messages.success(self.request, _("Shift copied successfully."))
        return super().form_valid(form)

    @classmethod
    def as_view(cls, **initkwargs):
        return csp_exempt(super().as_view(**initkwargs))

1	from copy import copy	1✔
2	from datetime import datetime, timedelta	1✔
3
4	from csp.decorators import csp_exempt	1✔
5	from django.contrib import messages	1✔
6	from django.core.exceptions import ValidationError	1✔
7	from django.shortcuts import get_object_or_404, redirect, render	1✔
8	from django.urls import reverse	1✔
9	from django.utils import timezone	1✔
10	from django.utils.datastructures import MultiValueDictKeyError	1✔
11	from django.utils.timezone import get_default_timezone	1✔
12	from django.utils.translation import gettext as _	1✔
13	from django.views import View	1✔
14	from django.views.generic import DeleteView, FormView, TemplateView	1✔
15	from django.views.generic.detail import SingleObjectMixin	1✔
16
17	from ephios.core.forms.events import ShiftCopyForm, ShiftForm	1✔
18	from ephios.core.models import Event, Shift	1✔
19	from ephios.core.signals import shift_forms	1✔
20	from ephios.core.signup.flow import enabled_signup_flows, signup_flow_from_slug	1✔
21	from ephios.core.signup.structure import enabled_shift_structures, shift_structure_from_slug	1✔
22	from ephios.extra.mixins import CustomPermissionRequiredMixin, PluginFormMixin	1✔
23
24
25	class ShiftCreateView(CustomPermissionRequiredMixin, PluginFormMixin, TemplateView):	1✔
26	permission_required = "core.change_event"	1✔
27	template_name = "core/shift_form.html"	1✔
28
29	def setup(self, request, args, *kwargs):	1✔
30	super().setup(request, args, *kwargs)	1✔
31	self.event = get_object_or_404(Event.all_objects, pk=self.kwargs.get("pk"))	1✔
32
33	def get_permission_object(self):	1✔
34	return self.event	1✔
35
36	def get_shift_form(self):	1✔
37	return ShiftForm(self.request.POST or None)	1✔
38
39	def get_context_data(self, **kwargs):	1✔
40	kwargs.setdefault("event", self.event)	1✔
41	kwargs.setdefault("form", self.get_shift_form())	1✔
42	kwargs.setdefault("flow_configuration_form", "")	1✔
43	kwargs.setdefault("structure_configuration_form", "")	1✔
44	return super().get_context_data(**kwargs)	1✔
45
46	def get_plugin_forms(self):	1✔
47	return shift_forms.send(	1✔
48	sender=None, shift=getattr(self, "object", None), request=self.request
49	)
50
51	def post(self, args, *kwargs):	1✔
52	form = self.get_shift_form()	1✔
53	self.object = form.instance	1✔
54
55	try:	1✔
56	signup_flow = signup_flow_from_slug(	1✔
57	self.request.POST.get("signup_flow_slug"), event=self.event
58	)
59	structure = shift_structure_from_slug(	1✔
60	self.request.POST.get("structure_slug"),
61	event=self.event,
62	)
63	except KeyError:	×
64	if not list(enabled_signup_flows()):	×
65	form.add_error(	×
66	"signup_flow_slug",
67	_("You must enable plugins providing signup flows to continue."),
68	)
69	if not list(enabled_shift_structures()):	×
70	form.add_error(	×
71	"structure_slug",
72	_("You must enable plugins providing shift structures to continue."),
73	)
74	return self.render_to_response(	×
75	self.get_context_data(
76	form=form,
77	)
78	)
79	except ValueError as e:	×
80	form.add_error(None, ValidationError(e))	×
81	return self.render_to_response(	×
82	self.get_context_data(
83	form=form,
84	)
85	)
86
87	flow_configuration_form = signup_flow.get_configuration_form(	1✔
88	self.request.POST, event=self.event, request=self.request
89	)
90	structure_configuration_form = structure.get_configuration_form(	1✔
91	self.request.POST, event=self.event, request=self.request
92	)
93	if not all(	1!
94	[
95	self.is_valid(form),
96	flow_configuration_form.is_valid(),
97	structure_configuration_form.is_valid(),
98	]
99	):
100	return self.render_to_response(	×
101	self.get_context_data(
102	form=form,
103	flow_configuration_form=flow_configuration_form,
104	structure_configuration_form=structure_configuration_form,
105	)
106	)
107
108	shift = form.save(commit=False)	1✔
109	shift.event = self.event	1✔
110	shift.signup_flow_configuration = flow_configuration_form.cleaned_data	1✔
111	shift.structure_configuration = structure_configuration_form.cleaned_data	1✔
112	shift.save()	1✔
113	self.save_plugin_forms()	1✔
114	if "addAnother" in self.request.POST:	1!
115	return redirect(reverse("core:event_createshift", kwargs={"pk": self.kwargs.get("pk")}))	×
116	if "copyShift" in self.request.POST:	1!
NEW 117	return redirect(reverse("core:shift_copy", kwargs={"pk": shift.pk}))	×
118	try:	1✔
119	self.event.activate()	1✔
120	messages.success(	1✔
121	self.request,
122	_("The event {title} has been saved.").format(title=self.event.title),
123	)
124	except ValidationError as e:	×
125	messages.error(self.request, e)	×
126	return redirect(self.event.get_absolute_url())	1✔
127
128
129	class AbstractShiftConfigurationFormView(CustomPermissionRequiredMixin, SingleObjectMixin, View):	1✔
130	queryset = Event.all_objects	1✔
131	permission_required = "core.change_event"	1✔
132	pk_url_kwarg = "event_id"	1✔
133
134	def get(self, request, args, *kwargs):	1✔
135	try:	×
136	shift = self.get_object().shifts.get(pk=request.GET.get("shift_id") or None)	×
137	except Shift.DoesNotExist:	×
138	shift = None	×
139	form = self.get_form(self.request, self.get_object(), shift=shift)	×
140	return render(	×
141	request,
142	"core/fragments/shift_signup_config_form.html",
143	{
144	"form": form,
145	},
146	)
147
148	def get_form(self, request, event, shift=None):	1✔
149	raise NotImplementedError	×
150
151
152	class SignupFlowConfigurationFormView(AbstractShiftConfigurationFormView):	1✔
153	def get_form(	1✔
154	self,
155	request,
156	event,
157	shift=None,
158	):
159	return signup_flow_from_slug(	×
160	self.kwargs.get("slug"), event=event, shift=shift
161	).get_configuration_form(request=request)
162
163
164	class ShiftStructureConfigurationFormView(AbstractShiftConfigurationFormView):	1✔
165	def get_form(self, request, event, shift=None):	1✔
166	return shift_structure_from_slug(	×
167	self.kwargs.get("slug"), event=event, shift=shift
168	).get_configuration_form(request=request)
169
170
171	class ShiftUpdateView(	1✔
172	CustomPermissionRequiredMixin, PluginFormMixin, SingleObjectMixin, TemplateView
173	):
174	model = Shift	1✔
175	template_name = "core/shift_form.html"	1✔
176	permission_required = "core.change_event"	1✔
177
178	def get_permission_object(self):	1✔
179	return self.get_object().event	1✔
180
181	def get_shift_form(self):	1✔
182	return ShiftForm(	1✔
183	self.request.POST or None,
184	instance=self.object,
185	initial={
186	"date": self.object.meeting_time.date(),
187	"meeting_time": self.object.meeting_time.astimezone(get_default_timezone()).time(),
188	"start_time": self.object.start_time.astimezone(get_default_timezone()).time(),
189	"end_time": self.object.end_time.astimezone(get_default_timezone()).time(),
190	},
191	)
192
193	def get_flow_configuration_form(self):	1✔
194	return self.object.signup_flow.get_configuration_form(	1✔
195	request=self.request, data=self.request.POST or None
196	)
197
198	def get_structure_configuration_form(self):	1✔
199	return self.object.structure.get_configuration_form(	1✔
200	request=self.request, data=self.request.POST or None
201	)
202
203	def get_context_data(self, **kwargs):	1✔
204	self.object = self.get_object()	1✔
205	kwargs.setdefault("event", self.object.event)	1✔
206	kwargs.setdefault("form", self.get_shift_form())	1✔
207	kwargs.setdefault("flow_configuration_form", self.get_flow_configuration_form())	1✔
208	kwargs.setdefault("structure_configuration_form", self.get_structure_configuration_form())	1✔
209	return super().get_context_data(**kwargs)	1✔
210
211	def get_plugin_forms(self):	1✔
212	return shift_forms.send(sender=None, shift=self.object, request=self.request)	1✔
213
214	def post(self, args, *kwargs):	1✔
215	self.object = self.get_object()	1✔
216	form = self.get_shift_form()	1✔
217	flow_configuration_form = self.get_flow_configuration_form()	1✔
218	structure_configuration_form = self.get_structure_configuration_form()	1✔
219	try:	1✔
220	signup_flow = signup_flow_from_slug(	1✔
221	self.request.POST["signup_flow_slug"], shift=self.object
222	)
223	flow_configuration_form = signup_flow.get_configuration_form(	1✔
224	self.request.POST,
225	event=self.object.event,
226	request=self.request,
227	)
228	structure = shift_structure_from_slug(	1✔
229	self.request.POST["structure_slug"], shift=self.object
230	)
231	structure_configuration_form = structure.get_configuration_form(	1✔
232	self.request.POST,
233	event=self.object.event,
234	request=self.request,
235	)
236	except (ValueError, MultiValueDictKeyError):	×
237	pass	×
238	else:
239	if all(	1✔
240	[
241	self.is_valid(form),
242	flow_configuration_form.is_valid(),
243	structure_configuration_form.is_valid(),
244	]
245	):
246	shift = form.save(commit=False)	1✔
247	shift.signup_flow_configuration = flow_configuration_form.cleaned_data	1✔
248	shift.structure_configuration = structure_configuration_form.cleaned_data	1✔
249	shift.save()	1✔
250	self.save_plugin_forms()	1✔
251	if "addAnother" in self.request.POST:	1!
252	return redirect(	×
253	reverse("core:event_createshift", kwargs={"pk": shift.event.pk})
254	)
255	if "copyShift" in self.request.POST:	1!
NEW 256	return redirect(reverse("core:shift_copy", kwargs={"pk": shift.pk}))	×
257	messages.success(	1✔
258	self.request, _("The shift {shift} has been saved.").format(shift=shift)
259	)
260	return redirect(self.object.event.get_absolute_url())	1✔
261
262	return self.render_to_response(	1✔
263	self.get_context_data(
264	form=form,
265	flow_configuration_form=flow_configuration_form,
266	structure_configuration_form=structure_configuration_form,
267	)
268	)
269
270
271	class ShiftDeleteView(CustomPermissionRequiredMixin, DeleteView):	1✔
272	permission_required = "core.change_event"	1✔
273	model = Shift	1✔
274
275	def setup(self, request, args, *kwargs):	1✔
276	super().setup(request, args, *kwargs)	1✔
277	self.object = self.get_object()	1✔
278
279	def form_valid(self, form):	1✔
280	if self.object.event.shifts.count() == 1:	1!
281	messages.error(self.request, _("You cannot delete the last shift!"))	×
282	return redirect(self.object.event.get_absolute_url())	×
283	return super().form_valid(form)	1✔
284
285	def get_success_url(self):	1✔
286	messages.success(self.request, _("The shift has been deleted."))	1✔
287	return self.object.event.get_absolute_url()	1✔
288
289	def get_permission_object(self):	1✔
290	return self.object.event	1✔
291
292
293	class ShiftCopyView(CustomPermissionRequiredMixin, SingleObjectMixin, FormView):	1✔
294	permission_required = "core.change_event"	1✔
295	model = Shift	1✔
296	template_name = "core/shift_copy.html"	1✔
297	form_class = ShiftCopyForm	1✔
298
299	def setup(self, request, args, *kwargs):	1✔
NEW 300	super().setup(request, args, *kwargs)	×
NEW 301	self.object = self.get_object()	×
302
303	def get_form_kwargs(self):	1✔
NEW 304	kwargs = super().get_form_kwargs()	×
NEW 305	kwargs["original_start"] = timezone.localtime(self.object.start_time).isoformat()	×
NEW 306	return kwargs	×
307
308	def get_success_url(self):	1✔
NEW 309	return reverse("core:event_detail", kwargs={"pk": self.object.event.pk, "slug": "event"})	×
310
311	def form_valid(self, form):	1✔
NEW 312	shift = self.object	×
NEW 313	duration = shift.end_time - shift.start_time	×
NEW 314	meeting_offset = shift.start_time - shift.meeting_time	×
NEW 315	shifts_to_create = []	×
NEW 316	recurr = form.cleaned_data["recurrence"]	×
NEW 317	tz = timezone.get_current_timezone()	×
NEW 318	for dt in recurr.xafter(datetime.now() - timedelta(days=365 * 100), 1000, inc=True):	×
NEW 319	dt = timezone.make_aware(dt, tz)	×
NEW 320	shift = copy(shift)	×
NEW 321	shift.pk = None	×
NEW 322	shift.meeting_time = dt - meeting_offset	×
NEW 323	shift.start_time = dt	×
NEW 324	shift.end_time = dt + duration	×
NEW 325	shifts_to_create.append(shift)	×
NEW 326	Shift.objects.bulk_create(shifts_to_create)	×
NEW 327	messages.success(self.request, _("Shift copied successfully."))	×
NEW 328	return super().form_valid(form)	×
329
330	@classmethod	1✔
331	def as_view(cls, **initkwargs):	1✔
332	return csp_exempt(super().as_view(**initkwargs))	1✔

ephios-dev / ephios / 12313892613

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