bika.lims.workflow.analysis.guards.guard_publish() - Code Metrics - Inspection of "Display "unassign" transition in Worksheet context..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( e9f2f3...ddf457 )

by Ramon

created 2019-01-28 09:44 UTC

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

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

Complexity

Conditions

Size

Total Lines	6
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	2
dl	0
loc	6
rs	10
c	0
b	0
f	0
cc	1
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 is_worksheet_context():
    """Returns whether the current context from the request is a Worksheet
    """
    request = api.get_request()
    parents = request.get("PARENTS", [])
    portal_types_names = map(lambda p: getattr(p, "portal_type", None), parents)
    if "Worksheet" in portal_types_names:
        return True

    # 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 IWorksheet.providedBy(obj):
        return True

    return False


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
    """
    # Only if the request was done from worksheet context.
    if not is_worksheet_context():
        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
    """
    # Only if the request was done from worksheet context.
    if not is_worksheet_context():
        return False

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

senaite / senaite.core

Push — master ( e9f2f3...ddf457 )

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

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like