bika.lims.workflow.analysisrequest.guards.guard_sample() - Code Metrics - Inspection of "Sample cleanup (#1180)" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Branch — master (9edffc)

by Jordi

created 2019-01-04 23:53 UTC

guard_sample() A

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

Complexity

Conditions

Size

Total Lines	10
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	5
dl	0
loc	10
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.workflow import isTransitionAllowed

# States to be omitted in regular transitions
ANALYSIS_DETACHED_STATES = ['cancelled', 'rejected', 'retracted']


def guard_no_sampling_workflow(analysis_request):
    """Returns whether the transition "no_sampling_workflow" can be performed
    or not. Returns True when Sampling Workflow is not enabled in setup
    """
    return not analysis_request.getSamplingRequired()


def guard_to_be_sampled(analysis_request):
    """Returns whether the transition "to_be_sampled" can be performed or not.
    Returns True if Sampling Workflow is enabled for the analysis request
    """
    return analysis_request.getSamplingRequired()


def guard_schedule_sampling(analysis_request):
    """Return whether the transition "schedule_sampling" can be performed or not
    Returns True only when the schedule sampling workflow is enabled in setup.
    """
    return analysis_request.bika_setup.getScheduleSamplingEnabled()


def guard_create_partitions(analysis_request):
    """Returns true if partitions can be created using the analysis request
    passed in as the source.
    """
    if not analysis_request.bika_setup.getShowPartitions():
        # If partitions are disabled in Setup, return False
        return False

    if analysis_request.isPartition():
        # Do not allow the creation of partitions from partitions
        return False

    # Allow only the creation of partitions if all analyses from the Analysis
    # Request are in unassigned state. Otherwise, we could end up with
    # inconsistencies, because original analyses are deleted when the partition
    # is created. Note here we exclude analyses from children (partitions).
    analyses = analysis_request.objectValues("Analysis")
    for analysis in analyses:
        if api.get_workflow_status_of(analysis) != "unassigned":
            return False
    return analyses and True or False


def guard_submit(analysis_request):
    """Return whether the transition "submit" can be performed or not.
    Returns True if there is at least one analysis in a non-detached state and
    all analyses in a non-detached analyses have been submitted.
    """
    analyses_ready = False
    for analysis in analysis_request.getAnalyses():
        analysis = api.get_object(analysis)
        analysis_status = api.get_workflow_status_of(analysis)
        if analysis_status in ANALYSIS_DETACHED_STATES:
            continue
        if analysis_status in ['assigned', 'unassigned']:
            return False
        analyses_ready = True
    return analyses_ready


def guard_verify(analysis_request):
    """Returns whether the transition "verify" can be performed or not.
    Returns True if at there is at least one analysis in a non-dettached state
    and all analyses in a non-dettached state are in "verified" state.
    """
    analyses_ready = False
    for analysis in analysis_request.getAnalyses():
        analysis = api.get_object(analysis)
        analysis_status = api.get_workflow_status_of(analysis)
        if analysis_status in ANALYSIS_DETACHED_STATES:
            continue
        if analysis_status != 'verified':
            return False
        analyses_ready = True
    return analyses_ready


def guard_prepublish(analysis_request):
    """Returns whether 'prepublish' transition can be perform or not. Returns
    True if the analysis request has at least one analysis in 'verified' or in
    'to_be_verified' status. Otherwise, return False
    """
    valid_states = ['verified', 'to_be_verified']
    for analysis in analysis_request.getAnalyses():
        analysis = api.get_object(analysis)
        if api.get_workflow_status_of(analysis) in valid_states:
            return True
    return False


def guard_rollback_to_receive(analysis_request):
    """Return whether 'rollback_to_receive' transition can be performed or not.
    Returns True if the analysis request has at least one analysis in 'assigned'
    or 'unassigned' status. Otherwise, returns False
    """
    skipped = 0
    valid_states = ['unassigned', 'assigned']
    skip_states = ['retracted', 'rejected']
    analyses = analysis_request.getAnalyses()
    for analysis in analyses:
        analysis = api.get_object(analysis)
        status = api.get_workflow_status_of(analysis)
        if status in valid_states:
            return True
        elif status in skip_states:
            skipped += 1
    return len(analyses) == skipped


def guard_cancel(analysis_request):
    """Returns whether 'cancel' transition can be performed or not. Returns
    True only if all analyses are in "unassigned" status
    """
    # Ask to partitions
    for partition in analysis_request.getDescendants(all_descendants=False):
        if not isTransitionAllowed(partition, "cancel"):
            return False

    # Look through analyses. We've checked the partitions already, so there is
    # no need to look through analyses from partitions again, but through the
    # analyses directly bound to the current Analysis Request.
    for analysis in analysis_request.objectValues("Analysis"):
        if api.get_workflow_status_of(analysis) != "unassigned":
            return False

    return True


def guard_reinstate(analysis_request):
    """Returns whether 'reinstate" transition can be performed or not. Returns
    True only if this is not a partition or the parent analysis request can be
    reinstated or is not in a cancelled state
    """
    parent = analysis_request.getParentAnalysisRequest()
    if not parent:
        return True
    if api.get_workflow_status_of(parent) != "cancelled":
        return True
    return isTransitionAllowed(parent, "reinstate")


def guard_sample(analysis_request):
    """Returns whether 'sample' transition can be performed or not. Returns
    True only if the analysis request has the DateSampled and Sampler set or if
    the user belongs to the Samplers group
    """
    if analysis_request.getDateSampled() and analysis_request.getSampler():
        return True

    current_user = api.get_current_user()
    return "Sampler" in current_user.getRolesInContext(analysis_request)


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.workflow import isTransitionAllowed
10
11			# States to be omitted in regular transitions
12			ANALYSIS_DETACHED_STATES = ['cancelled', 'rejected', 'retracted']
13
14
15			def guard_no_sampling_workflow(analysis_request):
16			"""Returns whether the transition "no_sampling_workflow" can be performed
17			or not. Returns True when Sampling Workflow is not enabled in setup
18			"""
19			return not analysis_request.getSamplingRequired()
20
21
22			def guard_to_be_sampled(analysis_request):
23			"""Returns whether the transition "to_be_sampled" can be performed or not.
24			Returns True if Sampling Workflow is enabled for the analysis request
25			"""
26			return analysis_request.getSamplingRequired()
27
28
29			def guard_schedule_sampling(analysis_request):
30			"""Return whether the transition "schedule_sampling" can be performed or not
31			Returns True only when the schedule sampling workflow is enabled in setup.
32			"""
33			return analysis_request.bika_setup.getScheduleSamplingEnabled()
34
35
36			def guard_create_partitions(analysis_request):
37			"""Returns true if partitions can be created using the analysis request
38			passed in as the source.
39			"""
40			if not analysis_request.bika_setup.getShowPartitions():
41			# If partitions are disabled in Setup, return False
42			return False
43
44			if analysis_request.isPartition():
45			# Do not allow the creation of partitions from partitions
46			return False
47
48			# Allow only the creation of partitions if all analyses from the Analysis
49			# Request are in unassigned state. Otherwise, we could end up with
50			# inconsistencies, because original analyses are deleted when the partition
51			# is created. Note here we exclude analyses from children (partitions).
52			analyses = analysis_request.objectValues("Analysis")
53			for analysis in analyses:
54			if api.get_workflow_status_of(analysis) != "unassigned":
55			return False
56			return analyses and True or False
57
58
59			def guard_submit(analysis_request):
60			"""Return whether the transition "submit" can be performed or not.
61			Returns True if there is at least one analysis in a non-detached state and
62			all analyses in a non-detached analyses have been submitted.
63			"""
64			analyses_ready = False
65			for analysis in analysis_request.getAnalyses():
66			analysis = api.get_object(analysis)
67			analysis_status = api.get_workflow_status_of(analysis)
68			if analysis_status in ANALYSIS_DETACHED_STATES:
69			continue
70			if analysis_status in ['assigned', 'unassigned']:
71			return False
72			analyses_ready = True
73			return analyses_ready
74
75
76			def guard_verify(analysis_request):
77			"""Returns whether the transition "verify" can be performed or not.
78			Returns True if at there is at least one analysis in a non-dettached state
79			and all analyses in a non-dettached state are in "verified" state.
80			"""
81			analyses_ready = False
82			for analysis in analysis_request.getAnalyses():
83			analysis = api.get_object(analysis)
84			analysis_status = api.get_workflow_status_of(analysis)
85			if analysis_status in ANALYSIS_DETACHED_STATES:
86			continue
87			if analysis_status != 'verified':
88			return False
89			analyses_ready = True
90			return analyses_ready
91
92
93			def guard_prepublish(analysis_request):
94			"""Returns whether 'prepublish' transition can be perform or not. Returns
95			True if the analysis request has at least one analysis in 'verified' or in
96			'to_be_verified' status. Otherwise, return False
97			"""
98			valid_states = ['verified', 'to_be_verified']
99			for analysis in analysis_request.getAnalyses():
100			analysis = api.get_object(analysis)
101			if api.get_workflow_status_of(analysis) in valid_states:
102			return True
103			return False
104
105
106			def guard_rollback_to_receive(analysis_request):
107			"""Return whether 'rollback_to_receive' transition can be performed or not.
108			Returns True if the analysis request has at least one analysis in 'assigned'
109			or 'unassigned' status. Otherwise, returns False
110			"""
111			skipped = 0
112			valid_states = ['unassigned', 'assigned']
113			skip_states = ['retracted', 'rejected']
114			analyses = analysis_request.getAnalyses()
115			for analysis in analyses:
116			analysis = api.get_object(analysis)
117			status = api.get_workflow_status_of(analysis)
118			if status in valid_states:
119			return True
120			elif status in skip_states:
121			skipped += 1
122			return len(analyses) == skipped
123
124
125			def guard_cancel(analysis_request):
126			"""Returns whether 'cancel' transition can be performed or not. Returns
127			True only if all analyses are in "unassigned" status
128			"""
129			# Ask to partitions
130			for partition in analysis_request.getDescendants(all_descendants=False):
131			if not isTransitionAllowed(partition, "cancel"):
132			return False
133
134			# Look through analyses. We've checked the partitions already, so there is
135			# no need to look through analyses from partitions again, but through the
136			# analyses directly bound to the current Analysis Request.
137			for analysis in analysis_request.objectValues("Analysis"):
138			if api.get_workflow_status_of(analysis) != "unassigned":
139			return False
140
141			return True
142
143
144			def guard_reinstate(analysis_request):
145			"""Returns whether 'reinstate" transition can be performed or not. Returns
146			True only if this is not a partition or the parent analysis request can be
147			reinstated or is not in a cancelled state
148			"""
149			parent = analysis_request.getParentAnalysisRequest()
150			if not parent:
151			return True
152			if api.get_workflow_status_of(parent) != "cancelled":
153			return True
154			return isTransitionAllowed(parent, "reinstate")
155
156
157			def guard_sample(analysis_request):
158			"""Returns whether 'sample' transition can be performed or not. Returns
159			True only if the analysis request has the DateSampled and Sampler set or if
160			the user belongs to the Samplers group
161			"""
162			if analysis_request.getDateSampled() and analysis_request.getSampler():
163			return True
164
165			current_user = api.get_current_user()
166			return "Sampler" in current_user.getRolesInContext(analysis_request)
167

senaite / senaite.core

Branch — master (9edffc)

guard_sample() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like