2422919867

Build # 2422919867

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/HewlettPackard/squest/commit/<a class=hub.com/HewlettPackard/squest/commit/<a class="double-link" href="https://git"><a class=hub.com/HewlettPackard/squest/commit/<a class="double-link" href="https://git"><a class=hub.com/HewlettPackard/squest/commit/<a class="double-link" href="https://git"><a class=hub.com/HewlettPackard/squest/commit/<a class="double-link" href="https://git"><a class=hub.com/HewlettPackard/squest/commit/ef6405e9dfc353db1703412b9322f6a45653a300">ef6405e9d">&lt;a href=&quot;https://github.com/HewlettPackard/squest/commit/</a><a class="double-link" href="https://github.com/HewlettPackard/squest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/HewlettPackard/squest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/HewlettPackard/squest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/HewlettPackard/squest/commit/ef6405e9dfc353db1703412b9322f6a45653a300">ef6405e9d</a><a href="https://github.com/HewlettPackard/squest/commit/ef6405e9dfc353db1703412b9322f6a45653a300">&lt;a href=&quot;https://github.com/HewlettPackard/squest/commit/ef6405e9dfc353db1703412b9322f6a45653a300&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/HewlettPackard/squest/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/HewlettPackard/squest/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/HewlettPackard/squest/commit/ef6405e9dfc353db1703412b9322f6a45653a300&amp;quot;&amp;gt;ef6405e9d&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/HewlettPackard/squest/commit/ef6405e9dfc353db1703412b9322f6a45653a300&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/HewlettPackard/squest/commit/ef6405e9dfc353db1703412b9322f6a4&amp;lt;/a&amp;gt;5653a300&amp;</a>quot;&gt;&amp;quot;&amp;gt;Merge &amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/HewlettPackard/squest/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/HewlettPackard/squest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/HewlettPackard/squ... (continued)

Pull Request Pull Request #445: [Feature] Approval group

Run Details

645 of 645 new or added lines in 50 files covered. (100.0%)

6999 of 7480 relevant lines covered (93.57%)

0.94 hits per line

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

97.78

/service_catalog/models/approval_step.py

from django.contrib.auth.models import User
from django.core.exceptions import ValidationError
from django.db.models import ForeignKey, ManyToManyField, CharField, SET_NULL, QuerySet, CASCADE, IntegerField
from django.db.models.signals import post_save, m2m_changed
from django.dispatch import receiver

from profiles.models import RoleManager
from profiles.models.team import Team
from service_catalog.models.approval_step_type import ApprovalStepType
from django.utils.translation import ugettext_lazy as _

from service_catalog.models.approval_state import ApprovalState


class ApprovalStep(RoleManager):
    class Meta:
        unique_together = (('id', 'approval_workflow'), ('name', 'approval_workflow'))

    approval_workflow = ForeignKey(
        "service_catalog.ApprovalWorkflow",
        blank=False,
        null=False,
        on_delete=CASCADE,
        related_name="approval_step_list"
    )

    name = CharField(max_length=100, blank=False)

    teams = ManyToManyField(
        Team,
        related_name='approval_steps',
        related_query_name='approval_step'
    )
    type = CharField(
        max_length=12,
        choices=ApprovalStepType.choices,
        blank=False,
        null=False,
    )
    next = ForeignKey(
        "service_catalog.ApprovalStep",
        blank=True,
        null=True,
        default=None,
        on_delete=SET_NULL,
        related_name="previous"
    )

    position = IntegerField(null=True, blank=True)

    @property
    def is_entry_point(self):
        return self.approval_workflow.entry_point == self

    def __str__(self):
        return self.name

    def get_approvers(self):
        queryset = User.objects.none()
        for team in self.teams.all():
            queryset = QuerySet.union(queryset, team.get_all_users())
        return queryset

    def get_approvers_emails(self):
        return [user.email for user in self.get_approvers()]

    def _assert_no_loop(self):
        workflow_list = list()
        current = self
        while current:
            if current in workflow_list:
                raise ValidationError({'next': _("You cannot create a loop.")})
            workflow_list.append(current)
            current = current.next

    def _assert_no_duplicate_name_in_the_workflow(self):
        if ApprovalStep.objects.exclude(id=self.id).filter(approval_workflow=self.approval_workflow,
                                                           name=self.name).exists():
            raise ValidationError({'name': _("The name already exist in the workflow.")})

    def clean(self):
        self._assert_no_loop()
        self._assert_no_duplicate_name_in_the_workflow()

    def set_next(self, approval_step_id=None):
        self.next = ApprovalStep.objects.get(id=approval_step_id) if approval_step_id is not None else None
        self.save()

    def get_approval_state(self, request):
        """
        Check the status of the current step
        """
        approval_state_list = self.approval_step_states.filter(request=request)
        state_list = [approval_state.state for approval_state in approval_state_list]
        if ApprovalState.REJECTED in state_list:
            return ApprovalState.REJECTED
        elif self.type == ApprovalStepType.ALL_OF_THEM:
            if state_list.count(ApprovalState.APPROVED) == len(state_list):
                return ApprovalState.APPROVED
            else:
                return ApprovalState.PENDING
        elif self.type == ApprovalStepType.AT_LEAST_ONE:
            if ApprovalState.APPROVED in state_list:
                return ApprovalState.APPROVED
            else:
                return ApprovalState.PENDING

    def delete(self, using=None, keep_parents=False):
        """
        Relink steps process when one of them is deleted
        """
        if self.approval_workflow.entry_point == self:
            self.approval_workflow.entry_point = self.next
            self.approval_workflow.save()
        previous = self.previous.first()
        if previous:
            previous.set_next(self.next.id if self.next else None)
        super(ApprovalStep, self).delete(using, keep_parents)
        self.approval_workflow.update_positions()

    def save(self, force_insert=False, force_update=False, using=None, update_fields=None):
        super(ApprovalStep, self).save(force_insert, force_update, using, update_fields)
        self.approval_workflow.update_positions()


@receiver(post_save, sender=ApprovalStep)
def set_approver_role_and_entrypoint_approval_workflow(sender, instance, created, **kwargs):
    """
    Add the Approver role and change the entry point of the approval workflow when needed
    """
    if created:
        for team in instance.teams.all():
            instance.add_team_in_role(team, "Approver")
        if not instance.approval_workflow.entry_point or instance.next == instance.approval_workflow.entry_point:
            instance.approval_workflow.entry_point = instance
            instance.approval_workflow.save()


def teams_changed(sender, instance, **kwargs):
    current_teams = instance.get_teams_in_role("Approver")
    to_remove = list(set(current_teams) - set(instance.teams.all()))
    to_add = list(set(instance.teams.all()) - set(current_teams))
    for team in to_add:
        instance.add_team_in_role(team, "Approver")
    for team in to_remove:
        instance.remove_team_in_role(team, "Approver")


m2m_changed.connect(teams_changed, sender=ApprovalStep.teams.through)

1	from django.contrib.auth.models import User	1✔
2	from django.core.exceptions import ValidationError	1✔
3	from django.db.models import ForeignKey, ManyToManyField, CharField, SET_NULL, QuerySet, CASCADE, IntegerField	1✔
4	from django.db.models.signals import post_save, m2m_changed	1✔
5	from django.dispatch import receiver	1✔
6
7	from profiles.models import RoleManager	1✔
8	from profiles.models.team import Team	1✔
9	from service_catalog.models.approval_step_type import ApprovalStepType	1✔
10	from django.utils.translation import ugettext_lazy as _	1✔
11
12	from service_catalog.models.approval_state import ApprovalState	1✔
13
14
15	class ApprovalStep(RoleManager):	1✔
16	class Meta:	1✔
17	unique_together = (('id', 'approval_workflow'), ('name', 'approval_workflow'))	1✔
18
19	approval_workflow = ForeignKey(	1✔
20	"service_catalog.ApprovalWorkflow",
21	blank=False,
22	null=False,
23	on_delete=CASCADE,
24	related_name="approval_step_list"
25	)
26
27	name = CharField(max_length=100, blank=False)	1✔
28
29	teams = ManyToManyField(	1✔
30	Team,
31	related_name='approval_steps',
32	related_query_name='approval_step'
33	)
34	type = CharField(	1✔
35	max_length=12,
36	choices=ApprovalStepType.choices,
37	blank=False,
38	null=False,
39	)
40	next = ForeignKey(	1✔
41	"service_catalog.ApprovalStep",
42	blank=True,
43	null=True,
44	default=None,
45	on_delete=SET_NULL,
46	related_name="previous"
47	)
48
49	position = IntegerField(null=True, blank=True)	1✔
50
51	@property	1✔
52	def is_entry_point(self):	1✔
53	return self.approval_workflow.entry_point == self	1✔
54
55	def __str__(self):	1✔
56	return self.name	1✔
57
58	def get_approvers(self):	1✔
59	queryset = User.objects.none()	1✔
60	for team in self.teams.all():	1✔
61	queryset = QuerySet.union(queryset, team.get_all_users())	1✔
62	return queryset	1✔
63
64	def get_approvers_emails(self):	1✔
65	return [user.email for user in self.get_approvers()]	1✔
66
67	def _assert_no_loop(self):	1✔
68	workflow_list = list()	1✔
69	current = self	1✔
70	while current:	1✔
71	if current in workflow_list:	1✔
72	raise ValidationError({'next': _("You cannot create a loop.")})	1✔
73	workflow_list.append(current)	1✔
74	current = current.next	1✔
75
76	def _assert_no_duplicate_name_in_the_workflow(self):	1✔
77	if ApprovalStep.objects.exclude(id=self.id).filter(approval_workflow=self.approval_workflow,	1✔
78	name=self.name).exists():
79	raise ValidationError({'name': _("The name already exist in the workflow.")})	1✔
80
81	def clean(self):	1✔
82	self._assert_no_loop()	1✔
83	self._assert_no_duplicate_name_in_the_workflow()	1✔
84
85	def set_next(self, approval_step_id=None):	1✔
86	self.next = ApprovalStep.objects.get(id=approval_step_id) if approval_step_id is not None else None	1✔
87	self.save()	1✔
88
89	def get_approval_state(self, request):	1✔
90	"""
91	Check the status of the current step
92	"""
93	approval_state_list = self.approval_step_states.filter(request=request)	1✔
94	state_list = [approval_state.state for approval_state in approval_state_list]	1✔
95	if ApprovalState.REJECTED in state_list:	1✔
96	return ApprovalState.REJECTED	1✔
97	elif self.type == ApprovalStepType.ALL_OF_THEM:	1✔
98	if state_list.count(ApprovalState.APPROVED) == len(state_list):	1✔
99	return ApprovalState.APPROVED	1✔
100	else:
101	return ApprovalState.PENDING	1✔
102	elif self.type == ApprovalStepType.AT_LEAST_ONE:	1✔
103	if ApprovalState.APPROVED in state_list:	1✔
104	return ApprovalState.APPROVED	1✔
105	else:
106	return ApprovalState.PENDING	×
107
108	def delete(self, using=None, keep_parents=False):	1✔
109	"""
110	Relink steps process when one of them is deleted
111	"""
112	if self.approval_workflow.entry_point == self:	1✔
113	self.approval_workflow.entry_point = self.next	1✔
114	self.approval_workflow.save()	1✔
115	previous = self.previous.first()	1✔
116	if previous:	1✔
117	previous.set_next(self.next.id if self.next else None)	1✔
118	super(ApprovalStep, self).delete(using, keep_parents)	1✔
119	self.approval_workflow.update_positions()	1✔
120
121	def save(self, force_insert=False, force_update=False, using=None, update_fields=None):	1✔
122	super(ApprovalStep, self).save(force_insert, force_update, using, update_fields)	1✔
123	self.approval_workflow.update_positions()	1✔
124
125
126	@receiver(post_save, sender=ApprovalStep)	1✔
127	def set_approver_role_and_entrypoint_approval_workflow(sender, instance, created, **kwargs):	1✔
128	"""
129	Add the Approver role and change the entry point of the approval workflow when needed
130	"""
131	if created:	1✔
132	for team in instance.teams.all():	1✔
133	instance.add_team_in_role(team, "Approver")	×
134	if not instance.approval_workflow.entry_point or instance.next == instance.approval_workflow.entry_point:	1✔
135	instance.approval_workflow.entry_point = instance	1✔
136	instance.approval_workflow.save()	1✔
137
138
139	def teams_changed(sender, instance, **kwargs):	1✔
140	current_teams = instance.get_teams_in_role("Approver")	1✔
141	to_remove = list(set(current_teams) - set(instance.teams.all()))	1✔
142	to_add = list(set(instance.teams.all()) - set(current_teams))	1✔
143	for team in to_add:	1✔
144	instance.add_team_in_role(team, "Approver")	1✔
145	for team in to_remove:	1✔
146	instance.remove_team_in_role(team, "Approver")	1✔
147
148
149	m2m_changed.connect(teams_changed, sender=ApprovalStep.teams.through)	1✔

HewlettPackard / squest / 2422919867

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