bika.lims.workflow.analysis.guards.guard_initialize() - Code Metrics - Inspection of "Disallow entry of analysis results if the sample i..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 5f146e...28e0c7 )

by Jordi

created 2019-01-27 19:52 UTC

guard_initialize() A

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

Complexity

Conditions

Size

Total Lines	7
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	5
dl	0
loc	7
rs	10
c	0
b	0
f	0
cc	2
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 import IWorksheet
from bika.lims.interfaces.analysis import IRequestAnalysis
from bika.lims import workflow as wf


def guard_initialize(analysis):
    """Return whether the transition "initialize" can be performed or not
    """
    request = analysis.getRequest()
    if request.getDateReceived():
        return True
    return False


def guard_assign(analysis):
    """Return whether the transition "assign" can be performed or not
    """

    # TODO: Refactor this funtion to a more generic place
    # only if the request was done from worksheet context.
    request = api.get_request()
    parents = request.get("PARENTS", [])
    portal_types_names = map(lambda p: getattr(p, "portal_type", None), parents)
    if "Worksheet" not in portal_types_names:
        # Check if the worksheet is declared in request explicitly
        ws_uid = request.get("ws_uid", "")
        obj = api.get_object_by_uid(ws_uid, None)
        if not IWorksheet.providedBy(obj):
            return False

    # 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
    if not user_can_manage_worksheets():
        return False

    return True


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

senaite / senaite.core

Push — master ( 5f146e...28e0c7 )

guard_initialize() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like