11443886939

Committed 21 Oct 2024 03:51PM UTC coverage: 85.621% (-0.04%) from 85.661%

Build # 11443886939

Build Type

Pull #1388

github

Committed by

web-flow

Commit Message

Merge bf44f6292 into 0609b0ec3

Pull Request Pull Request #1388: Fix teams structure fullness display

Run Details

2929 of 3427 branches covered (85.47%)

Branch coverage included in aggregate %.

22 of 28 new or added lines in 2 files covered. (78.57%)

1 existing line in 1 file now uncovered.

11576 of 13514 relevant lines covered (85.66%)

0.86 hits per line

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

85.66

/ephios/plugins/baseshiftstructures/structure/named_teams.py

import uuid
from collections import Counter
from functools import cached_property
from itertools import groupby
from operator import itemgetter

from django import forms
from django.core.exceptions import ValidationError
from django.utils.translation import gettext_lazy as _
from django.utils.translation import ngettext_lazy

from ephios.core.models import AbstractParticipation, Qualification
from ephios.core.signup.disposition import BaseDispositionParticipationForm
from ephios.core.signup.flow.participant_validation import ParticipantUnfitError
from ephios.core.signup.forms import BaseSignupForm
from ephios.core.signup.participants import AbstractParticipant
from ephios.plugins.baseshiftstructures.structure.group_common import (
    AbstractGroupBasedStructureConfigurationForm,
    BaseGroupBasedShiftStructure,
    QualificationRequirementForm,
    format_min_max_count,
)


def teams_participant_qualifies_for(teams, participant: AbstractParticipant):
    available_qualification_ids = set(q.id for q in participant.collect_all_qualifications())
    return [
        team
        for team in teams
        if not (q := team.get("qualification")) or q in available_qualification_ids
    ]


class NamedTeamsDispositionParticipationForm(BaseDispositionParticipationForm):
    disposition_participation_template = (
        "baseshiftstructures/named_teams/fragment_participation.html"
    )

    team = forms.ChoiceField(
        label=_("Section"),
        required=False,  # only required if participation is confirmed
        widget=forms.Select(
            attrs={"data-show-for-state": str(AbstractParticipation.States.CONFIRMED)}
        ),
    )

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        teams = self.shift.structure.configuration.teams
        qualified_teams = list(
            teams_participant_qualifies_for(
                teams,
                self.instance.participant,
            )
        )
        unqualified_teams = [team for team in teams if team not in qualified_teams]
        self.fields["team"].choices = [("", "---")]
        if qualified_teams:
            self.fields["team"].choices += [
                (
                    _("qualified"),
                    [(team["uuid"], team["title"]) for team in qualified_teams],
                )
            ]
        if unqualified_teams:
            self.fields["team"].choices += [
                (
                    _("unqualified"),
                    [(team["uuid"], team["title"]) for team in unqualified_teams],
                )
            ]
        if preferred_team_uuid := self.instance.structure_data.get("preferred_team_uuid"):
            self.fields["team"].initial = preferred_team_uuid
            self.preferred_team = next(
                filter(lambda team: team["uuid"] == preferred_team_uuid, teams), None
            )
        if initial := self.instance.structure_data.get("dispatched_team_uuid"):
            self.fields["team"].initial = initial

    def clean(self):
        super().clean()
        if (
            self.cleaned_data["state"] == AbstractParticipation.States.CONFIRMED
            and not self.cleaned_data["team"]
        ):
            self.add_error(
                "team",
                ValidationError(_("You must select a team when confirming a participation.")),
            )

    def save(self, commit=True):
        self.instance.structure_data["dispatched_team_uuid"] = self.cleaned_data["team"]
        super().save(commit)


class NamedTeamsSignupForm(BaseSignupForm):
    preferred_team_uuid = forms.ChoiceField(
        label=_("Preferred Team"),
        widget=forms.RadioSelect,
        required=False,
        # choices are later set as (uuid, title) of team
    )

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.fields["preferred_team_uuid"].initial = self.instance.structure_data.get(
            "preferred_team_uuid"
        )
        self.fields["preferred_team_uuid"].required = (
            self.data.get("signup_choice") == "sign_up"
            and self.shift.structure.configuration.choose_preferred_team
        )

        team_stats = self.shift.structure._get_signup_stats_per_group(
            self.shift.participations.all()
        )
        enabled_teams = []
        not_qualified_teams = []
        full_teams = []
        for team in self.shift.structure.configuration.teams:
            if team["uuid"] in [
                self.instance.structure_data.get("preferred_team_uuid"),
                self.instance.structure_data.get("dispatched_team_uuid"),
            ]:
                enabled_teams.append(team)
            elif team not in self.teams_participant_qualifies_for:
                not_qualified_teams.append(team)
            elif (
                not self.shift.signup_flow.uses_requested_state
                and not team_stats[team["uuid"]].has_free()
            ):
                full_teams.append(team)
            else:
                enabled_teams.append(team)

        self.fields["preferred_team_uuid"].choices = [
            (team["uuid"], team["title"]) for team in enabled_teams
        ]
        help_text = ""
        if not_qualified_teams:
            help_text = _("You don't qualify for {teams}.").format(
                teams=", ".join(str(team["title"]) for team in not_qualified_teams)
            )
        if full_teams:
            help_text += " " + ngettext_lazy(
                "{teams} is full.", "{teams} are full.", len(full_teams)
            ).format(teams=", ".join(str(team["title"]) for team in full_teams))
        if help_text:
            self.fields["preferred_team_uuid"].help_text = help_text

    def save(self, commit=True):
        self.instance.structure_data["preferred_team_uuid"] = self.cleaned_data[
            "preferred_team_uuid"
        ]
        return super().save(commit)

    @cached_property
    def teams_participant_qualifies_for(self):
        return teams_participant_qualifies_for(
            self.shift.structure.configuration.teams, self.participant
        )


class NamedTeamForm(QualificationRequirementForm):
    title = forms.CharField(label=_("Title"), required=True)
    uuid = forms.CharField(widget=forms.HiddenInput, required=False)

    def clean_uuid(self):
        return self.cleaned_data.get("uuid") or uuid.uuid4()


NamedTeamsFormset = forms.formset_factory(
    NamedTeamForm, can_delete=True, min_num=1, validate_min=1, extra=0
)


class NamedTeamsConfigurationForm(AbstractGroupBasedStructureConfigurationForm):
    template_name = "baseshiftstructures/named_teams/configuration_form.html"
    choose_preferred_team = forms.BooleanField(
        label=_("Participants must provide a preferred team"),
        help_text=_(
            "Participants will be asked during signup. This only makes sense if you configure multiple teams."
        ),
        widget=forms.CheckboxInput,
        required=False,
        initial=False,
    )
    teams = forms.Field(
        label=_("Teams"),
        widget=forms.HiddenInput,
        required=False,
    )
    formset_data_field_name = "teams"

    def get_formset_class(self):
        return NamedTeamsFormset

    @classmethod
    def format_formset_item(cls, item):
        return item["title"]


class NamedTeamsShiftStructure(BaseGroupBasedShiftStructure):
    slug = "named_teams"
    verbose_name = _("Named teams")
    description = _("Define named teams of participants with different requirements.")
    configuration_form_class = NamedTeamsConfigurationForm
    shift_state_template_name = "baseshiftstructures/named_teams/fragment_state.html"
    disposition_participation_form_class = NamedTeamsDispositionParticipationForm
    signup_form_class = NamedTeamsSignupForm

    NO_TEAM_UUID = "noteam"

    def _choose_team_for_participation(self, participation):
        return participation.structure_data.get(
            "dispatched_team_uuid"
        ) or participation.structure_data.get("preferred_team_uuid")

    def _get_signup_stats_per_group(self, participations):
        from ephios.core.signup.stats import SignupStats

        confirmed_counter = Counter()
        requested_counter = Counter()
        for p in participations:
            if p.state == AbstractParticipation.States.CONFIRMED:
                c = confirmed_counter
            elif p.state == AbstractParticipation.States.REQUESTED:
                c = requested_counter
            else:
                continue
            team_uuid = self._choose_team_for_participation(p)
            if not team_uuid or team_uuid not in (
                team["uuid"] for team in self.configuration.teams
            ):
                team_uuid = self.NO_TEAM_UUID
            c[team_uuid] += 1

        d = {}
        for team in self.configuration.teams:
            team_uuid = team["uuid"]
            min_count = team.get("min_count")
            max_count = team.get("max_count")
            d[team_uuid] = SignupStats(
                requested_count=requested_counter[team_uuid],
                confirmed_count=confirmed_counter[team_uuid],
                missing=(max(min_count - confirmed_counter[team_uuid], 0) if min_count else 0),
                free=(max(max_count - confirmed_counter[team_uuid], 0) if max_count else None),
                min_count=min_count,
                max_count=max_count,
            )

        # Participations not assigned to a team are extra, so max and free are explicitly zero.
        # We do not offset missing places in other teams, as qualifications etc. might not match.
        # Disposition will always be required to resolve unassigned participations.
        d[self.NO_TEAM_UUID] = SignupStats(
            requested_count=requested_counter[self.NO_TEAM_UUID],
            confirmed_count=confirmed_counter[self.NO_TEAM_UUID],
            missing=0,
            free=0,
            min_count=None,
            max_count=0,
        )

        return d

    def get_checkers(self):
        def check_qualifications_and_max_count(shift, participant):
            viable_teams = teams_participant_qualifies_for(
                shift.structure.configuration.teams, participant
            )
            if not viable_teams:
                raise ParticipantUnfitError(_("You are not qualified."))

            # check if teams are full if signup flow does not use requested state
            if shift.signup_flow.uses_requested_state:
                return
            free_team = False
            team_stats = self._get_signup_stats_per_group(self.shift.participations.all())
            for team in viable_teams:
                if team_stats[team["uuid"]].has_free():
                    free_team = True
                    break
            if not free_team:
                raise ParticipantUnfitError(_("All teams you qualify for are full."))

        return super().get_checkers() + [check_qualifications_and_max_count]

    def _configure_participation(
        self, participation: AbstractParticipation, **kwargs
    ) -> AbstractParticipation:
        participation.state = AbstractParticipation.States.REQUESTED
        return participation

    def get_shift_state_context_data(self, request, **kwargs):
        context_data = super().get_shift_state_context_data(request)
        participations = context_data["participations"]
        teams_stats = self._get_signup_stats_per_group(participations)
        teams = {}
        for team in self.configuration.teams:
            try:
                qualification = Qualification.objects.get(id=team["qualification"])
            except Qualification.DoesNotExist:
                qualification = None
            teams[team["uuid"]] = {
                "title": team["title"],
                "placeholder": team.get("min_count") or 0,
                "qualification_label": qualification.abbreviation if qualification else "",
                "min_max_count": format_min_max_count(team.get("min_count"), team.get("max_count")),
                "participations": [],
                "stats": teams_stats[team["uuid"]],
            }

        unsorted_participations = []
        for participation in participations:
            dispatched_uuid = self._choose_team_for_participation(participation)
            if dispatched_uuid not in teams:
                unsorted_participations.append(participation)
            else:
                teams[dispatched_uuid]["participations"].append(participation)
                teams[dispatched_uuid]["placeholder"] -= 1

        for team in teams.values():
            team["placeholder"] = list(range(max(0, team["placeholder"])))
        if unsorted_participations:
            teams[self.NO_TEAM_UUID] = {
                "title": _("unassigned"),
                "participations": unsorted_participations,
                "placeholder": [],
                "stats": teams_stats[self.NO_TEAM_UUID],
            }

        context_data["teams"] = teams
        return context_data

    def _get_teams_with_users(self):
        team_by_uuid = {team["uuid"]: team for team in self.configuration.teams}
        # get name and preferred team uuid for confirmed participants
        # if they have a team assigned and we have that team on record
        confirmed_participations = [
            {
                "name": str(participation.participant),
                "relevant_qualifications": ", ".join(
                    participation.participant.qualifications.filter(
                        category__show_with_user=True,
                    )
                    .order_by("category", "abbreviation")
                    .values_list("abbreviation", flat=True)
                ),
                "uuid": dispatched_team_uuid,
            }
            for participation in self.shift.participations.filter(
                state=AbstractParticipation.States.CONFIRMED
            )
            if (dispatched_team_uuid := participation.structure_data.get("dispatched_team_uuid"))
            and dispatched_team_uuid in team_by_uuid
        ]
        # group by team and do some stats
        teams_with_users = [
            (
                team_by_uuid.pop(uuid),
                [[user["name"], user["relevant_qualifications"]] for user in group],
            )
            for uuid, group in groupby(
                sorted(confirmed_participations, key=itemgetter("uuid")), itemgetter("uuid")
            )
        ]
        # add teams without participants
        teams_with_users += [(team, None) for team in team_by_uuid.values()]
        return teams_with_users

    def get_participation_display(self):
        confirmed_teams_with_users = self._get_teams_with_users()
        participation_display = []
        for team, users in confirmed_teams_with_users:
            if users:
                participation_display += [[user[0], user[1], team["title"]] for user in users]
            if not users or len(users) < team["min_count"]:
                required_qualifications = ", ".join(
                    Qualification.objects.filter(pk__in=[team.get("qualification")]).values_list(
                        "abbreviation", flat=True
                    )
                )
                participation_display += [["", required_qualifications, team["title"]]] * (
                    team["min_count"] - (len(users) if users else 0)
                )
        return participation_display

1	import uuid	1✔
2	from collections import Counter	1✔
3	from functools import cached_property	1✔
4	from itertools import groupby	1✔
5	from operator import itemgetter	1✔
6
7	from django import forms	1✔
8	from django.core.exceptions import ValidationError	1✔
9	from django.utils.translation import gettext_lazy as _	1✔
10	from django.utils.translation import ngettext_lazy	1✔
11
12	from ephios.core.models import AbstractParticipation, Qualification	1✔
13	from ephios.core.signup.disposition import BaseDispositionParticipationForm	1✔
14	from ephios.core.signup.flow.participant_validation import ParticipantUnfitError	1✔
15	from ephios.core.signup.forms import BaseSignupForm	1✔
16	from ephios.core.signup.participants import AbstractParticipant	1✔
17	from ephios.plugins.baseshiftstructures.structure.group_common import (	1✔
18	AbstractGroupBasedStructureConfigurationForm,
19	BaseGroupBasedShiftStructure,
20	QualificationRequirementForm,
21	format_min_max_count,
22	)
23
24
25	def teams_participant_qualifies_for(teams, participant: AbstractParticipant):	1✔
26	available_qualification_ids = set(q.id for q in participant.collect_all_qualifications())	1✔
27	return [	1✔
28	team
29	for team in teams
30	if not (q := team.get("qualification")) or q in available_qualification_ids
31	]
32
33
34	class NamedTeamsDispositionParticipationForm(BaseDispositionParticipationForm):	1✔
35	disposition_participation_template = (	1✔
36	"baseshiftstructures/named_teams/fragment_participation.html"
37	)
38
39	team = forms.ChoiceField(	1✔
40	label=_("Section"),
41	required=False, # only required if participation is confirmed
42	widget=forms.Select(
43	attrs={"data-show-for-state": str(AbstractParticipation.States.CONFIRMED)}
44	),
45	)
46
47	def __init__(self, **kwargs):	1✔
48	super().__init__(**kwargs)	1✔
49	teams = self.shift.structure.configuration.teams	1✔
50	qualified_teams = list(	1✔
51	teams_participant_qualifies_for(
52	teams,
53	self.instance.participant,
54	)
55	)
56	unqualified_teams = [team for team in teams if team not in qualified_teams]	1✔
57	self.fields["team"].choices = [("", "---")]	1✔
58	if qualified_teams:	1!
59	self.fields["team"].choices += [	1✔
60	(
61	_("qualified"),
62	[(team["uuid"], team["title"]) for team in qualified_teams],
63	)
64	]
65	if unqualified_teams:	1!
66	self.fields["team"].choices += [	1✔
67	(
68	_("unqualified"),
69	[(team["uuid"], team["title"]) for team in unqualified_teams],
70	)
71	]
72	if preferred_team_uuid := self.instance.structure_data.get("preferred_team_uuid"):	1!
73	self.fields["team"].initial = preferred_team_uuid	1✔
74	self.preferred_team = next(	1✔
75	filter(lambda team: team["uuid"] == preferred_team_uuid, teams), None
76	)
77	if initial := self.instance.structure_data.get("dispatched_team_uuid"):	1!
78	self.fields["team"].initial = initial	×
79
80	def clean(self):	1✔
81	super().clean()	1✔
82	if (	1!
83	self.cleaned_data["state"] == AbstractParticipation.States.CONFIRMED
84	and not self.cleaned_data["team"]
85	):
86	self.add_error(	×
87	"team",
88	ValidationError(_("You must select a team when confirming a participation.")),
89	)
90
91	def save(self, commit=True):	1✔
92	self.instance.structure_data["dispatched_team_uuid"] = self.cleaned_data["team"]	1✔
93	super().save(commit)	1✔
94
95
96	class NamedTeamsSignupForm(BaseSignupForm):	1✔
97	preferred_team_uuid = forms.ChoiceField(	1✔
98	label=_("Preferred Team"),
99	widget=forms.RadioSelect,
100	required=False,
101	# choices are later set as (uuid, title) of team
102	)
103
104	def __init__(self, args, *kwargs):	1✔
105	super().__init__(args, *kwargs)	1✔
106	self.fields["preferred_team_uuid"].initial = self.instance.structure_data.get(	1✔
107	"preferred_team_uuid"
108	)
109	self.fields["preferred_team_uuid"].required = (	1✔
110	self.data.get("signup_choice") == "sign_up"
111	and self.shift.structure.configuration.choose_preferred_team
112	)
113
114	team_stats = self.shift.structure._get_signup_stats_per_group(	1✔
115	self.shift.participations.all()
116	)
117	enabled_teams = []	1✔
118	not_qualified_teams = []	1✔
119	full_teams = []	1✔
120	for team in self.shift.structure.configuration.teams:	1✔
121	if team["uuid"] in [	1!
122	self.instance.structure_data.get("preferred_team_uuid"),
123	self.instance.structure_data.get("dispatched_team_uuid"),
124	]:
NEW 125	enabled_teams.append(team)	×
126	elif team not in self.teams_participant_qualifies_for:	1✔
127	not_qualified_teams.append(team)	1✔
128	elif (	1!
129	not self.shift.signup_flow.uses_requested_state
130	and not team_stats[team["uuid"]].has_free()
131	):
NEW 132	full_teams.append(team)	×
133	else:
134	enabled_teams.append(team)	1✔
135
136	self.fields["preferred_team_uuid"].choices = [	1✔
137	(team["uuid"], team["title"]) for team in enabled_teams
138	]
139	help_text = ""	1✔
140	if not_qualified_teams:	1!
141	help_text = _("You don't qualify for {teams}.").format(	1✔
142	teams=", ".join(str(team["title"]) for team in not_qualified_teams)
143	)
144	if full_teams:	1!
NEW 145	help_text += " " + ngettext_lazy(	×
146	"{teams} is full.", "{teams} are full.", len(full_teams)
147	).format(teams=", ".join(str(team["title"]) for team in full_teams))
148	if help_text:	1!
149	self.fields["preferred_team_uuid"].help_text = help_text	1✔
150
151	def save(self, commit=True):	1✔
152	self.instance.structure_data["preferred_team_uuid"] = self.cleaned_data[	1✔
153	"preferred_team_uuid"
154	]
155	return super().save(commit)	1✔
156
157	@cached_property	1✔
158	def teams_participant_qualifies_for(self):	1✔
159	return teams_participant_qualifies_for(	1✔
160	self.shift.structure.configuration.teams, self.participant
161	)
162
163
164	class NamedTeamForm(QualificationRequirementForm):	1✔
165	title = forms.CharField(label=_("Title"), required=True)	1✔
166	uuid = forms.CharField(widget=forms.HiddenInput, required=False)	1✔
167
168	def clean_uuid(self):	1✔
169	return self.cleaned_data.get("uuid") or uuid.uuid4()	1✔
170
171
172	NamedTeamsFormset = forms.formset_factory(	1✔
173	NamedTeamForm, can_delete=True, min_num=1, validate_min=1, extra=0
174	)
175
176
177	class NamedTeamsConfigurationForm(AbstractGroupBasedStructureConfigurationForm):	1✔
178	template_name = "baseshiftstructures/named_teams/configuration_form.html"	1✔
179	choose_preferred_team = forms.BooleanField(	1✔
180	label=_("Participants must provide a preferred team"),
181	help_text=_(
182	"Participants will be asked during signup. This only makes sense if you configure multiple teams."
183	),
184	widget=forms.CheckboxInput,
185	required=False,
186	initial=False,
187	)
188	teams = forms.Field(	1✔
189	label=_("Teams"),
190	widget=forms.HiddenInput,
191	required=False,
192	)
193	formset_data_field_name = "teams"	1✔
194
195	def get_formset_class(self):	1✔
196	return NamedTeamsFormset	1✔
197
198	@classmethod	1✔
199	def format_formset_item(cls, item):	1✔
200	return item["title"]	1✔
201
202
203	class NamedTeamsShiftStructure(BaseGroupBasedShiftStructure):	1✔
204	slug = "named_teams"	1✔
205	verbose_name = _("Named teams")	1✔
206	description = _("Define named teams of participants with different requirements.")	1✔
207	configuration_form_class = NamedTeamsConfigurationForm	1✔
208	shift_state_template_name = "baseshiftstructures/named_teams/fragment_state.html"	1✔
209	disposition_participation_form_class = NamedTeamsDispositionParticipationForm	1✔
210	signup_form_class = NamedTeamsSignupForm	1✔
211
212	NO_TEAM_UUID = "noteam"	1✔
213
214	def _choose_team_for_participation(self, participation):	1✔
215	return participation.structure_data.get(	1✔
216	"dispatched_team_uuid"
217	) or participation.structure_data.get("preferred_team_uuid")
218
219	def _get_signup_stats_per_group(self, participations):	1✔
220	from ephios.core.signup.stats import SignupStats	1✔
221
222	confirmed_counter = Counter()	1✔
223	requested_counter = Counter()	1✔
224	for p in participations:	1✔
225	if p.state == AbstractParticipation.States.CONFIRMED:	1✔
226	c = confirmed_counter	1✔
227	elif p.state == AbstractParticipation.States.REQUESTED:	1!
228	c = requested_counter	1✔
229	else:
230	continue	×
231	team_uuid = self._choose_team_for_participation(p)	1✔
232	if not team_uuid or team_uuid not in (	1!
233	team["uuid"] for team in self.configuration.teams
234	):
235	team_uuid = self.NO_TEAM_UUID	1✔
236	c[team_uuid] += 1	1✔
237
238	d = {}	1✔
239	for team in self.configuration.teams:	1✔
240	team_uuid = team["uuid"]	1✔
241	min_count = team.get("min_count")	1✔
242	max_count = team.get("max_count")	1✔
243	d[team_uuid] = SignupStats(	1✔
244	requested_count=requested_counter[team_uuid],
245	confirmed_count=confirmed_counter[team_uuid],
246	missing=(max(min_count - confirmed_counter[team_uuid], 0) if min_count else 0),
247	free=(max(max_count - confirmed_counter[team_uuid], 0) if max_count else None),
248	min_count=min_count,
249	max_count=max_count,
250	)
251
252	# Participations not assigned to a team are extra, so max and free are explicitly zero.
253	# We do not offset missing places in other teams, as qualifications etc. might not match.
254	# Disposition will always be required to resolve unassigned participations.
255	d[self.NO_TEAM_UUID] = SignupStats(	1✔
256	requested_count=requested_counter[self.NO_TEAM_UUID],
257	confirmed_count=confirmed_counter[self.NO_TEAM_UUID],
258	missing=0,
259	free=0,
260	min_count=None,
261	max_count=0,
262	)
263
264	return d	1✔
265
266	def get_checkers(self):	1✔
267	def check_qualifications_and_max_count(shift, participant):	1✔
268	viable_teams = teams_participant_qualifies_for(	1✔
269	shift.structure.configuration.teams, participant
270	)
271	if not viable_teams:	1!
272	raise ParticipantUnfitError(_("You are not qualified."))	×
273
274	# check if teams are full if signup flow does not use requested state
275	if shift.signup_flow.uses_requested_state:	1!
276	return	1✔
277	free_team = False	×
NEW 278	team_stats = self._get_signup_stats_per_group(self.shift.participations.all())	×
279	for team in viable_teams:	×
NEW 280	if team_stats[team["uuid"]].has_free():	×
UNCOV 281	free_team = True	×
282	break	×
283	if not free_team:	×
284	raise ParticipantUnfitError(_("All teams you qualify for are full."))	×
285
286	return super().get_checkers() + [check_qualifications_and_max_count]	1✔
287
288	def _configure_participation(	1✔
289	self, participation: AbstractParticipation, **kwargs
290	) -> AbstractParticipation:
291	participation.state = AbstractParticipation.States.REQUESTED	×
292	return participation	×
293
294	def get_shift_state_context_data(self, request, **kwargs):	1✔
295	context_data = super().get_shift_state_context_data(request)	1✔
296	participations = context_data["participations"]	1✔
297	teams_stats = self._get_signup_stats_per_group(participations)	1✔
298	teams = {}	1✔
299	for team in self.configuration.teams:	1✔
300	try:	1✔
301	qualification = Qualification.objects.get(id=team["qualification"])	1✔
302	except Qualification.DoesNotExist:	1✔
303	qualification = None	1✔
304	teams[team["uuid"]] = {	1✔
305	"title": team["title"],
306	"placeholder": team.get("min_count") or 0,
307	"qualification_label": qualification.abbreviation if qualification else "",
308	"min_max_count": format_min_max_count(team.get("min_count"), team.get("max_count")),
309	"participations": [],
310	"stats": teams_stats[team["uuid"]],
311	}
312
313	unsorted_participations = []	1✔
314	for participation in participations:	1✔
315	dispatched_uuid = self._choose_team_for_participation(participation)	1✔
316	if dispatched_uuid not in teams:	1✔
317	unsorted_participations.append(participation)	1✔
318	else:
319	teams[dispatched_uuid]["participations"].append(participation)	1✔
320	teams[dispatched_uuid]["placeholder"] -= 1	1✔
321
322	for team in teams.values():	1✔
323	team["placeholder"] = list(range(max(0, team["placeholder"])))	1✔
324	if unsorted_participations:	1✔
325	teams[self.NO_TEAM_UUID] = {	1✔
326	"title": _("unassigned"),
327	"participations": unsorted_participations,
328	"placeholder": [],
329	"stats": teams_stats[self.NO_TEAM_UUID],
330	}
331
332	context_data["teams"] = teams	1✔
333	return context_data	1✔
334
335	def _get_teams_with_users(self):	1✔
336	team_by_uuid = {team["uuid"]: team for team in self.configuration.teams}	1✔
337	# get name and preferred team uuid for confirmed participants
338	# if they have a team assigned and we have that team on record
339	confirmed_participations = [	1✔
340	{
341	"name": str(participation.participant),
342	"relevant_qualifications": ", ".join(
343	participation.participant.qualifications.filter(
344	category__show_with_user=True,
345	)
346	.order_by("category", "abbreviation")
347	.values_list("abbreviation", flat=True)
348	),
349	"uuid": dispatched_team_uuid,
350	}
351	for participation in self.shift.participations.filter(
352	state=AbstractParticipation.States.CONFIRMED
353	)
354	if (dispatched_team_uuid := participation.structure_data.get("dispatched_team_uuid"))
355	and dispatched_team_uuid in team_by_uuid
356	]
357	# group by team and do some stats
358	teams_with_users = [	1✔
359	(
360	team_by_uuid.pop(uuid),
361	[[user["name"], user["relevant_qualifications"]] for user in group],
362	)
363	for uuid, group in groupby(
364	sorted(confirmed_participations, key=itemgetter("uuid")), itemgetter("uuid")
365	)
366	]
367	# add teams without participants
368	teams_with_users += [(team, None) for team in team_by_uuid.values()]	1✔
369	return teams_with_users	1✔
370
371	def get_participation_display(self):	1✔
372	confirmed_teams_with_users = self._get_teams_with_users()	1✔
373	participation_display = []	1✔
374	for team, users in confirmed_teams_with_users:	1✔
375	if users:	1✔
376	participation_display += [[user[0], user[1], team["title"]] for user in users]	1✔
377	if not users or len(users) < team["min_count"]:	1!
378	required_qualifications = ", ".join(	1✔
379	Qualification.objects.filter(pk__in=[team.get("qualification")]).values_list(
380	"abbreviation", flat=True
381	)
382	)
383	participation_display += [["", required_qualifications, team["title"]]] * (	1✔
384	team["min_count"] - (len(users) if users else 0)
385	)
386	return participation_display	1✔

ephios-dev / ephios / 11443886939

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