bika.lims.workflow.analysis.guards.guard_assign() - Code Metrics - Inspection of "Remove attachment_due state from analysis workflow..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 743380...0e02ba )

by Ramon

created 2018-12-03 09:34 UTC

bika.lims.workflow.analysis.guards.guard_assign() A

↳ Parent: bika.lims.workflow.analysis.guards

Complexity

Conditions

Size

Total Lines	22
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	6
dl	0
loc	22
rs	10
c	0
b	0
f	0
cc	3
nop	1

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE
#
# Copyright 2018 by it's authors.
# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.

from bika.lims import api
from bika.lims.interfaces.analysis import IRequestAnalysis
from bika.lims import workflow as wf


def guard_assign(analysis):
    """Return whether the transition "assign" can be performed or not
    """
    # TODO Workflow - Analysis. Assign guard to return True only in WS.Add?
    #      We need "unassigned" analyses to appear in Worksheet Add analyses.
    #      Hence, it returns True if the analysis has not been assigned to any
    #      worksheet yet. The problem is this can end up (if the 'assign'
    #      transition is displayed in listings other than WS Add Analyses)
    #      with an analysis transitioned to 'assigned' state, but without
    #      a worksheet assigned!. This transition should only be triggered by
    #      content.worksheet.addAnalysis (see that func for more info)

    # Cannot assign if the Sample has not been received
    if not analysis.isSampleReceived():
        return False

    # Cannot assign if the analysis has a worksheet assigned already
    if analysis.getWorksheet():
        return False

    # Cannot assign if user does not have permissions to manage worksheets
    return user_can_manage_worksheets()


def guard_unassign(analysis):
    """Return whether the transition "unassign" can be performed or not
    """
    # Cannot unassign if the analysis is not assigned to any worksheet
    if not analysis.getWorksheet():
        return False

    # Cannot unassign if user does not have permissions to manage worksheets
    return user_can_manage_worksheets()


def guard_cancel(analysis):
    """Return whether the transition "cancel" can be performed or not. Returns
    True only when the Analysis Request the analysis belongs to is in cancelled
    state. Otherwise, returns False.
    """
    return not api.is_active(analysis.getRequest())


def guard_reinstate(analysis):
    """Return whether the transition "reinstate" can be performed or not.
    Returns True only when the Analysis Request the analysis belongs to is in a
    non-cancelled state. Otherwise, returns False.
    """
    return api.is_active(analysis.getRequest())


def guard_submit(analysis):
    """Return whether the transition "submit" can be performed or not
    """
    # Cannot submit without a result
    if not analysis.getResult():
        return False

    # Cannot submit with interims without value
    for interim in analysis.getInterimFields():
        if not interim.get("value", ""):
            return False

    # Cannot submit if attachment not set, but is required
    if not analysis.getAttachment():
        if analysis.getAttachmentOption() == 'r':
            return False

    # Check if can submit based on the Analysis Request state
    if IRequestAnalysis.providedBy(analysis):
        point_of_capture = analysis.getPointOfCapture()
        # Cannot submit if the Sample has not been received
        if point_of_capture == "lab" and not analysis.isSampleReceived():
            return False
        # Cannot submit if the Sample has not been sampled
        if point_of_capture == "field" and not analysis.isSampleSampled():
            return False

    # Check if the current user can submit if is not assigned
    if not analysis.bika_setup.getAllowToSubmitNotAssigned():
        if not user_has_super_roles():
            # Cannot submit if unassigned
            if not analysis.getAnalyst():
                return False
            # Cannot submit if assigned analyst is not the current user
            if analysis.getAnalyst() != api.get_current_user().getId():
                return False

    # Cannot submit unless all dependencies are submitted or can be submitted
    dependencies = analysis.getDependencies()
    return is_transition_allowed_or_performed(dependencies, "submit")


def guard_multi_verify(analysis):
    """Return whether the transition "multi_verify" can be performed or not
    The transition multi_verify will only take place if multi-verification of
    results is enabled.
    """
    # Cannot multiverify if there is only one remaining verification
    remaining_verifications = analysis.getNumberOfRemainingVerifications()
    if remaining_verifications <= 1:
        return False

    # Cannot verify if the user submitted and self-verification is not allowed
    if was_submitted_by_current_user(analysis):
        if not analysis.isSelfVerificationEnabled():
            return False

    # Cannot verify if the user verified and multi verification is not allowed
    if was_verified_by_current_user(analysis):
        if not is_multi_verification_allowed(analysis):
            return False

    # Cannot verify if the user was last verifier and consecutive verification
    # by same user is not allowed
    if current_user_was_last_verifier(analysis):
        if not is_consecutive_multi_verification_allowed(analysis):
            return False

    # Cannot verify unless all dependencies are verified or can be verified
    dependencies = analysis.getDependencies()
    transitions = ["verify", "multi_verify"]
    return is_transition_allowed_or_performed(dependencies, transitions)


def guard_verify(analysis):
    """Return whether the transition "verify" can be performed or not
    """
    # Cannot verify if the number of remaining verifications is > 1
    remaining_verifications = analysis.getNumberOfRemainingVerifications()
    if remaining_verifications > 1:
        return False

    # Cannot verify if the user submitted and self-verification is not allowed
    if was_submitted_by_current_user(analysis):
        if not analysis.isSelfVerificationEnabled():
            return False

    # Cannot verify unless dependencies have been verified or can be verified
    if analysis.getNumberOfRequiredVerifications() <= 1:
        dependencies = analysis.getDependencies()
        return is_transition_allowed_or_performed(dependencies, "verify")

    # This analysis has multi-verification enabled
    # Cannot verify if the user verified and multi verification is not allowed
    if was_verified_by_current_user(analysis):
        if not is_multi_verification_allowed(analysis):
            return False

    # Cannot verify if the user was last verifier and consecutive verification
    # by same user is not allowed
    if current_user_was_last_verifier(analysis):
        if not is_consecutive_multi_verification_allowed(analysis):
            return False

    # Cannot verify unless all dependencies are verified or can be verified
    dependencies = analysis.getDependencies()
    transitions = ["verify", "multi_verify"]
    return is_transition_allowed_or_performed(dependencies, transitions)


def guard_retract(analysis):
    """ Return whether the transition "retract" can be performed or not
    """
    # Cannot retract if there are dependents that cannot be retracted
    if not is_transition_allowed(analysis.getDependents(), "retract"):
        return False

    # Cannot retract if all dependencies have been verified
    return not was_transition_performed(analysis.getDependencies(), "verify")


def guard_reject(analysis):
    """Return whether the transition "reject" can be performed or not
    """
    # Cannot reject if there are dependents that cannot be rejected
    return is_transition_allowed(analysis.getDependents(), "reject")


def guard_publish(analysis):
    """Return whether the transition "publish" can be performed or not. Returns
    True only when the Analysis Request the analysis belongs to is in published
    state. Otherwise, returns False.
    """
    return api.get_workflow_status_of(analysis.getRequest()) == "published"


def user_can_manage_worksheets():
    """Return whether the current user has privileges to manage worksheets
    """
    if not api.get_setup().getRestrictWorksheetManagement():
        # There is no restriction, everybody can manage worksheets
        return True

    # Only Labmanager and Manager roles can manage worksheets
    return user_has_super_roles()


def user_has_super_roles():
    """Return whether the current belongs to superuser roles
    """
    member = api.get_current_user()
    super_roles = ["LabManager", "Manager"]
    diff = filter(lambda role: role in super_roles, member.getRoles())
    return len(diff) > 0


def was_submitted_by_current_user(analysis):
    """Returns whether the analysis was submitted by current user or not
    """
    return analysis.getSubmittedBy() == api.get_current_user().getId()


def was_verified_by_current_user(analysis):
    """Returns whether the analysis was verified by current user or not
    """
    return api.get_current_user().getId() in analysis.getVerificators()


def current_user_was_last_verifier(analysis):
    """Returns whether the current user was the last verifier or not
    """
    verifiers = analysis.getVerificators()
    return verifiers and verifiers[:-1] == api.get_current_user().getId()


def is_consecutive_multi_verification_allowed(analysis):
    """Returns whether multiple verification and consecutive verification is
    allowed or not"""
    multi_type = api.get_setup().getTypeOfmultiVerification()
    return multi_type != "self_multi_not_cons"


def is_multi_verification_allowed(analysis):
    """Returns whether multi verification is allowed or not
    """
    multi_type = api.get_setup().getTypeOfmultiVerification()
    return multi_type != "self_multi_disabled"


def is_transition_allowed(analyses, transition_id):
    """Returns whether all analyses can be transitioned or not
    """
    if not analyses:
        return True
    if not isinstance(analyses, list):
        return is_transition_allowed([analyses], transition_id)
    for analysis in analyses:
        if not wf.isTransitionAllowed(analysis, transition_id):
            return False
    return True


def was_transition_performed(analyses, transition_id):
    """Returns whether all analyses were transitioned or not
    """
    if not analyses:
        return False
    if not isinstance(analyses, list):
        return was_transition_performed([analyses], transition_id)
    for analysis in analyses:
        if not wf.wasTransitionPerformed(analysis, transition_id):
            return False
    return True


def is_transition_allowed_or_performed(analyses, transition_ids):
    """Return whether all analyses can be transitioned or all them were
    transitioned.
    """
    if not analyses:
        return True
    if not isinstance(analyses, list):
        return is_transition_allowed_or_performed([analyses], transition_ids)
    if not isinstance(transition_ids, list):
        return is_transition_allowed_or_performed(analyses, [transition_ids])

    for transition_id in transition_ids:
        for analysis in analyses:
            if not wf.isTransitionAllowed(analysis, transition_id):
                if not wf.wasTransitionPerformed(analysis, transition_id):
                    return False
    return True


1			# -- coding: utf-8 --
2			#
3			# This file is part of SENAITE.CORE
4			#
5			# Copyright 2018 by it's authors.
6			# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.
7
8			from bika.lims import api
9			from bika.lims.interfaces.analysis import IRequestAnalysis
10			from bika.lims import workflow as wf
11
12
13			def guard_assign(analysis):
14			"""Return whether the transition "assign" can be performed or not
15			"""
16			# TODO Workflow - Analysis. Assign guard to return True only in WS.Add?
17			# We need "unassigned" analyses to appear in Worksheet Add analyses.
18			# Hence, it returns True if the analysis has not been assigned to any
19			# worksheet yet. The problem is this can end up (if the 'assign'
20			# transition is displayed in listings other than WS Add Analyses)
21			# with an analysis transitioned to 'assigned' state, but without
22			# a worksheet assigned!. This transition should only be triggered by
23			# content.worksheet.addAnalysis (see that func for more info)
24
25			# Cannot assign if the Sample has not been received
26			if not analysis.isSampleReceived():
27			return False
28
29			# Cannot assign if the analysis has a worksheet assigned already
30			if analysis.getWorksheet():
31			return False
32
33			# Cannot assign if user does not have permissions to manage worksheets
34			return user_can_manage_worksheets()
35
36
37			def guard_unassign(analysis):
38			"""Return whether the transition "unassign" can be performed or not
39			"""
40			# Cannot unassign if the analysis is not assigned to any worksheet
41			if not analysis.getWorksheet():
42			return False
43
44			# Cannot unassign if user does not have permissions to manage worksheets
45			return user_can_manage_worksheets()
46
47
48			def guard_cancel(analysis):
49			"""Return whether the transition "cancel" can be performed or not. Returns
50			True only when the Analysis Request the analysis belongs to is in cancelled
51			state. Otherwise, returns False.
52			"""
53			return not api.is_active(analysis.getRequest())
54
55
56			def guard_reinstate(analysis):
57			"""Return whether the transition "reinstate" can be performed or not.
58			Returns True only when the Analysis Request the analysis belongs to is in a
59			non-cancelled state. Otherwise, returns False.
60			"""
61			return api.is_active(analysis.getRequest())
62
63
64			def guard_submit(analysis):
65			"""Return whether the transition "submit" can be performed or not
66			"""
67			# Cannot submit without a result
68			if not analysis.getResult():
69			return False
70
71			# Cannot submit with interims without value
72			for interim in analysis.getInterimFields():
73			if not interim.get("value", ""):
74			return False
75
76			# Cannot submit if attachment not set, but is required
77			if not analysis.getAttachment():
78			if analysis.getAttachmentOption() == 'r':
79			return False
80
81			# Check if can submit based on the Analysis Request state
82			if IRequestAnalysis.providedBy(analysis):
83			point_of_capture = analysis.getPointOfCapture()
84			# Cannot submit if the Sample has not been received
85			if point_of_capture == "lab" and not analysis.isSampleReceived():
86			return False
87			# Cannot submit if the Sample has not been sampled
88			if point_of_capture == "field" and not analysis.isSampleSampled():
89			return False
90
91			# Check if the current user can submit if is not assigned
92			if not analysis.bika_setup.getAllowToSubmitNotAssigned():
93			if not user_has_super_roles():
94			# Cannot submit if unassigned
95			if not analysis.getAnalyst():
96			return False
97			# Cannot submit if assigned analyst is not the current user
98			if analysis.getAnalyst() != api.get_current_user().getId():
99			return False
100
101			# Cannot submit unless all dependencies are submitted or can be submitted
102			dependencies = analysis.getDependencies()
103			return is_transition_allowed_or_performed(dependencies, "submit")
104
105
106			def guard_multi_verify(analysis):
107			"""Return whether the transition "multi_verify" can be performed or not
108			The transition multi_verify will only take place if multi-verification of
109			results is enabled.
110			"""
111			# Cannot multiverify if there is only one remaining verification
112			remaining_verifications = analysis.getNumberOfRemainingVerifications()
113			if remaining_verifications <= 1:
114			return False
115
116			# Cannot verify if the user submitted and self-verification is not allowed
117			if was_submitted_by_current_user(analysis):
118			if not analysis.isSelfVerificationEnabled():
119			return False
120
121			# Cannot verify if the user verified and multi verification is not allowed
122			if was_verified_by_current_user(analysis):
123			if not is_multi_verification_allowed(analysis):
124			return False
125
126			# Cannot verify if the user was last verifier and consecutive verification
127			# by same user is not allowed
128			if current_user_was_last_verifier(analysis):
129			if not is_consecutive_multi_verification_allowed(analysis):
130			return False
131
132			# Cannot verify unless all dependencies are verified or can be verified
133			dependencies = analysis.getDependencies()
134			transitions = ["verify", "multi_verify"]
135			return is_transition_allowed_or_performed(dependencies, transitions)
136
137
138			def guard_verify(analysis):
139			"""Return whether the transition "verify" can be performed or not
140			"""
141			# Cannot verify if the number of remaining verifications is > 1
142			remaining_verifications = analysis.getNumberOfRemainingVerifications()
143			if remaining_verifications > 1:
144			return False
145
146			# Cannot verify if the user submitted and self-verification is not allowed
147			if was_submitted_by_current_user(analysis):
148			if not analysis.isSelfVerificationEnabled():
149			return False
150
151			# Cannot verify unless dependencies have been verified or can be verified
152			if analysis.getNumberOfRequiredVerifications() <= 1:
153			dependencies = analysis.getDependencies()
154			return is_transition_allowed_or_performed(dependencies, "verify")
155
156			# This analysis has multi-verification enabled
157			# Cannot verify if the user verified and multi verification is not allowed
158			if was_verified_by_current_user(analysis):
159			if not is_multi_verification_allowed(analysis):
160			return False
161
162			# Cannot verify if the user was last verifier and consecutive verification
163			# by same user is not allowed
164			if current_user_was_last_verifier(analysis):
165			if not is_consecutive_multi_verification_allowed(analysis):
166			return False
167
168			# Cannot verify unless all dependencies are verified or can be verified
169			dependencies = analysis.getDependencies()
170			transitions = ["verify", "multi_verify"]
171			return is_transition_allowed_or_performed(dependencies, transitions)
172
173
174			def guard_retract(analysis):
175			""" Return whether the transition "retract" can be performed or not
176			"""
177			# Cannot retract if there are dependents that cannot be retracted
178			if not is_transition_allowed(analysis.getDependents(), "retract"):
179			return False
180
181			# Cannot retract if all dependencies have been verified
182			return not was_transition_performed(analysis.getDependencies(), "verify")
183
184
185			def guard_reject(analysis):
186			"""Return whether the transition "reject" can be performed or not
187			"""
188			# Cannot reject if there are dependents that cannot be rejected
189			return is_transition_allowed(analysis.getDependents(), "reject")
190
191
192			def guard_publish(analysis):
193			"""Return whether the transition "publish" can be performed or not. Returns
194			True only when the Analysis Request the analysis belongs to is in published
195			state. Otherwise, returns False.
196			"""
197			return api.get_workflow_status_of(analysis.getRequest()) == "published"
198
199
200			def user_can_manage_worksheets():
201			"""Return whether the current user has privileges to manage worksheets
202			"""
203			if not api.get_setup().getRestrictWorksheetManagement():
204			# There is no restriction, everybody can manage worksheets
205			return True
206
207			# Only Labmanager and Manager roles can manage worksheets
208			return user_has_super_roles()
209
210
211			def user_has_super_roles():
212			"""Return whether the current belongs to superuser roles
213			"""
214			member = api.get_current_user()
215			super_roles = ["LabManager", "Manager"]
216			diff = filter(lambda role: role in super_roles, member.getRoles())
217			return len(diff) > 0
218
219
220			def was_submitted_by_current_user(analysis):
221			"""Returns whether the analysis was submitted by current user or not
222			"""
223			return analysis.getSubmittedBy() == api.get_current_user().getId()
224
225
226			def was_verified_by_current_user(analysis):
227			"""Returns whether the analysis was verified by current user or not
228			"""
229			return api.get_current_user().getId() in analysis.getVerificators()
230
231
232			def current_user_was_last_verifier(analysis):
233			"""Returns whether the current user was the last verifier or not
234			"""
235			verifiers = analysis.getVerificators()
236			return verifiers and verifiers[:-1] == api.get_current_user().getId()
237
238
239			def is_consecutive_multi_verification_allowed(analysis):
240			"""Returns whether multiple verification and consecutive verification is
241			allowed or not"""
242			multi_type = api.get_setup().getTypeOfmultiVerification()
243			return multi_type != "self_multi_not_cons"
244
245
246			def is_multi_verification_allowed(analysis):
247			"""Returns whether multi verification is allowed or not
248			"""
249			multi_type = api.get_setup().getTypeOfmultiVerification()
250			return multi_type != "self_multi_disabled"
251
252
253			def is_transition_allowed(analyses, transition_id):
254			"""Returns whether all analyses can be transitioned or not
255			"""
256			if not analyses:
257			return True
258			if not isinstance(analyses, list):
259			return is_transition_allowed([analyses], transition_id)
260			for analysis in analyses:
261			if not wf.isTransitionAllowed(analysis, transition_id):
262			return False
263			return True
264
265
266			def was_transition_performed(analyses, transition_id):
267			"""Returns whether all analyses were transitioned or not
268			"""
269			if not analyses:
270			return False
271			if not isinstance(analyses, list):
272			return was_transition_performed([analyses], transition_id)
273			for analysis in analyses:
274			if not wf.wasTransitionPerformed(analysis, transition_id):
275			return False
276			return True
277
278
279			def is_transition_allowed_or_performed(analyses, transition_ids):
280			"""Return whether all analyses can be transitioned or all them were
281			transitioned.
282			"""
283			if not analyses:
284			return True
285			if not isinstance(analyses, list):
286			return is_transition_allowed_or_performed([analyses], transition_ids)
287			if not isinstance(transition_ids, list):
288			return is_transition_allowed_or_performed(analyses, [transition_ids])
289
290			for transition_id in transition_ids:
291			for analysis in analyses:
292			if not wf.isTransitionAllowed(analysis, transition_id):
293			if not wf.wasTransitionPerformed(analysis, transition_id):
294			return False
295			return True
296

senaite / senaite.core

Push — master ( 743380...0e02ba )

bika.lims.workflow.analysis.guards.guard_assign() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like