bika.lims.workflow.analysis.events.after_retract() - Code Metrics - Inspection of "Primary AR does not recognize created Partitions (..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 42e1f6...ef1ceb )

by Ramon

created 2018-11-16 09:42 UTC

bika.lims.workflow.analysis.events.after_retract() A

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

Complexity

Conditions

Size

Total Lines	55
Code Lines	23

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	23
dl	0
loc	55
rs	9.328
c	0
b	0
f	0
cc	4
nop	1

How to fix Long Method

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

import transaction
from Products.CMFCore.utils import getToolByName
from bika.lims.interfaces import IDuplicateAnalysis

from bika.lims.interfaces.analysis import IRequestAnalysis
from bika.lims.utils import changeWorkflowState
from bika.lims.utils.analysis import create_analysis
from bika.lims.workflow import doActionFor
from bika.lims.workflow import skip


def after_submit(obj):
    """Method triggered after a 'submit' transition for the analysis passed in
    is performed. Promotes the submit transition to the Worksheet to which the
    analysis belongs to. Note that for the worksheet there is already a guard
    that assures the transition to the worksheet will only be performed if all
    analyses within the worksheet have already been transitioned.
    This function is called automatically by
    bika.lims.workfow.AfterTransitionEventHandler
    """
    ws = obj.getWorksheet()
    if ws:
        doActionFor(ws, 'submit')

    if IRequestAnalysis.providedBy(obj):
        ar = obj.getRequest()
        doActionFor(ar, 'submit')
        reindex_request(obj)


def after_retract(obj):
    """Function triggered after a 'retract' transition for the analysis passed
    in is performed. Retracting an analysis cause its transition to 'retracted'
    state and the creation of a new copy of the same analysis as a retest.
    Note that retraction only affects to single Analysis and has no other
    effect in the status of the Worksheet to which the Analysis is assigned or
    to the Analysis Request to which belongs (transition is never proomoted)
    This function is called automatically by
    bika.lims.workflow.AfterTransitionEventHandler
    """
    # TODO Workflow Analysis - review this function
    # Rename the analysis to make way for it's successor.
    # Support multiple retractions by renaming to *-0, *-1, etc
    parent = obj.aq_parent
    kw = obj.getKeyword()
    analyses = [x for x in parent.objectValues("Analysis")
                if x.getId().startswith(obj.getId())]

    # LIMS-1290 - Analyst must be able to retract, which creates a new
    # Analysis.  So, _verifyObjectPaste permission check must be cancelled:
    parent._verifyObjectPaste = str
    # This is needed for tests:
    # https://docs.plone.org/develop/plone/content/rename.html
    # Testing warning: Rename mechanism relies of Persistent attribute
    # called _p_jar to be present on the content object. By default, this is
    # not the case on unit tests. You need to call transaction.savepoint() to
    # make _p_jar appear on persistent objects.
    # If you don't do this, you'll receive a "CopyError" when calling
    # manage_renameObjects that the operation is not supported.
    transaction.savepoint()
    parent.manage_renameObject(kw, "{0}-{1}".format(kw, len(analyses)))
    delattr(parent, '_verifyObjectPaste')

    # Create new analysis from the retracted obj
    analysis = create_analysis(parent, obj)
    changeWorkflowState(
        analysis, "bika_analysis_workflow", "sample_received")

    # Assign the new analysis to this same worksheet, if any.
    ws = obj.getWorksheet()
    if ws:
        ws.addAnalysis(analysis)
    analysis.reindexObject()

    # retract our dependencies
    dependencies = obj.getDependencies()
    for dependency in dependencies:
        doActionFor(dependency, 'retract')

    # Retract our dependents
    dependents = obj.getDependents()
    for dependent in dependents:
        doActionFor(dependent, 'retract')

    reindex_request(obj)


def after_verify(obj):
    """
    Method triggered after a 'verify' transition for the analysis passed in
    is performed. Promotes the transition to the Analysis Request and to
    Worksheet (if the analysis is assigned to any)
    This function is called automatically by
    bika.lims.workfow.AfterTransitionEventHandler
    """

    # If the analysis has dependencies, transition them
    for dependency in obj.getDependencies():
        doActionFor(dependency, 'verify')

    # Do all the reflex rules process
    obj._reflex_rule_process('verify')

    # Escalate to Analysis Request. Note that the guard for verify transition
    # from Analysis Request will check if the AR can be transitioned, so there
    # is no need to check here if all analyses within the AR have been
    # transitioned already.
    ar = obj.getRequest()
    doActionFor(ar, 'verify')

    # Ecalate to Worksheet. Note that the guard for verify transition from
    # Worksheet will check if the Worksheet can be transitioned, so there is no
    # need to check here if all analyses within the WS have been transitioned
    # already
    ws = obj.getWorksheet()
    if ws:
        doActionFor(ws, 'verify')
    reindex_request(obj)


def after_assign(obj):
    """Function triggered after an 'assign' transition for the analysis passed
    in is performed."""
    # Reindex the entire request to update the FieldIndex `assigned_state`
    reindex_request(obj, idxs=['assigned_state',])


def after_unassign(obj):
    """Function triggered after an 'unassign' transition for the analysis passed
    in is performed."""
    # Reindex the entire request to update the FieldIndex `assigned_state`
    reindex_request(obj, idxs=['assigned_state',])


def after_cancel(obj):
    if skip(obj, "cancel"):
        return
    workflow = getToolByName(obj, "portal_workflow")
    # If it is assigned to a worksheet, unassign it.
    state = workflow.getInfoFor(obj, 'worksheetanalysis_review_state')
    if state == 'assigned':
        ws = obj.getWorksheet()
        skip(obj, "cancel", unskip=True)
        ws.removeAnalysis(obj)
    obj.reindexObject()
    reindex_request(obj)


def after_reject(obj):
    if skip(obj, "reject"):
        return
    workflow = getToolByName(obj, "portal_workflow")
    # If it is assigned to a worksheet, unassign it.
    state = workflow.getInfoFor(obj, 'worksheetanalysis_review_state')
    if state == 'assigned':
        ws = obj.getWorksheet()
        ws.removeAnalysis(obj)
    obj.reindexObject()
    reindex_request(obj)


def after_attach(obj):
    if skip(obj, "attach"):
        return
    workflow = getToolByName(obj, "portal_workflow")
    # If all analyses in this AR have been attached escalate the action
    # to the parent AR
    ar = obj.aq_parent
    state = workflow.getInfoFor(ar, "review_state")
    if state == "attachment_due" and not skip(ar, "attach", peek=True):
        can_attach = True
        for a in ar.getAnalyses():
            if a.review_state in ("to_be_sampled", "to_be_preserved",
                                  "sample_due", "sample_received",
                                  "attachment_due"):
                can_attach = False
                break
        if can_attach:
            workflow.doActionFor(ar, "attach")
    # If assigned to a worksheet and all analyses on the worksheet have
    # been attached, then attach the worksheet.
    ws = obj.getBackReferences('WorksheetAnalysis')
    if ws:
        ws_state = workflow.getInfoFor(ws, "review_state")
        if ws_state == "attachment_due" \

                and not skip(ws, "attach", peek=True):
            can_attach = True
            for a in ws.getAnalyses():
                state = workflow.getInfoFor(a, "review_state")
                if state in ("to_be_sampled", "to_be_preserved",
                             "sample_due", "sample_received",
                             "attachment_due", "assigned"):
                    can_attach = False
                    break
            if can_attach:
                workflow.doActionFor(ws, "attach")
    obj.reindexObject()
    reindex_request(obj)


# TODO Workflow - Analysis - revisit reindexing of ancestors
def reindex_request(analysis, idxs=None):
    """Reindex the Analysis Request the analysis belongs to, as well as the
    ancestors recursively
    """
    if not IRequestAnalysis.providedBy(analysis) or \
            IDuplicateAnalysis.providedBy(analysis):
        # Analysis not directly bound to an Analysis Request. Do nothing
        return

    n_idxs = ['assigned_state', 'isRootAncestor', 'getDueDate']
    n_idxs = idxs and list(set(idxs + n_idxs)) or n_idxs
    analysis_request = analysis.getRequest()
    analysis_request.reindexObject(idxs=n_idxs)
    ancestors = analysis_request.getAncestors(all_ancestors=True)
    for ancestor in ancestors:
        ancestor.reindexObject(idxs=n_idxs)


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		import transaction
9		from Products.CMFCore.utils import getToolByName
10		from bika.lims.interfaces import IDuplicateAnalysis
11
12		from bika.lims.interfaces.analysis import IRequestAnalysis
13		from bika.lims.utils import changeWorkflowState
14		from bika.lims.utils.analysis import create_analysis
15		from bika.lims.workflow import doActionFor
16		from bika.lims.workflow import skip
17
18
19		def after_submit(obj):
20		"""Method triggered after a 'submit' transition for the analysis passed in
21		is performed. Promotes the submit transition to the Worksheet to which the
22		analysis belongs to. Note that for the worksheet there is already a guard
23		that assures the transition to the worksheet will only be performed if all
24		analyses within the worksheet have already been transitioned.
25		This function is called automatically by
26		bika.lims.workfow.AfterTransitionEventHandler
27		"""
28		ws = obj.getWorksheet()
29		if ws:
30		doActionFor(ws, 'submit')
31
32		if IRequestAnalysis.providedBy(obj):
33		ar = obj.getRequest()
34		doActionFor(ar, 'submit')
35		reindex_request(obj)
36
37
38		def after_retract(obj):
39		"""Function triggered after a 'retract' transition for the analysis passed
40		in is performed. Retracting an analysis cause its transition to 'retracted'
41		state and the creation of a new copy of the same analysis as a retest.
42		Note that retraction only affects to single Analysis and has no other
43		effect in the status of the Worksheet to which the Analysis is assigned or
44		to the Analysis Request to which belongs (transition is never proomoted)
45		This function is called automatically by
46		bika.lims.workflow.AfterTransitionEventHandler
47		"""
48		# TODO Workflow Analysis - review this function
49		# Rename the analysis to make way for it's successor.
50		# Support multiple retractions by renaming to -0, -1, etc
51		parent = obj.aq_parent
52		kw = obj.getKeyword()
53		analyses = [x for x in parent.objectValues("Analysis")
54		if x.getId().startswith(obj.getId())]
55
56		# LIMS-1290 - Analyst must be able to retract, which creates a new
57		# Analysis. So, _verifyObjectPaste permission check must be cancelled:
58		parent._verifyObjectPaste = str
59		# This is needed for tests:
60		# https://docs.plone.org/develop/plone/content/rename.html
61		# Testing warning: Rename mechanism relies of Persistent attribute
62		# called _p_jar to be present on the content object. By default, this is
63		# not the case on unit tests. You need to call transaction.savepoint() to
64		# make _p_jar appear on persistent objects.
65		# If you don't do this, you'll receive a "CopyError" when calling
66		# manage_renameObjects that the operation is not supported.
67		transaction.savepoint()
68		parent.manage_renameObject(kw, "{0}-{1}".format(kw, len(analyses)))
69		delattr(parent, '_verifyObjectPaste')
70
71		# Create new analysis from the retracted obj
72		analysis = create_analysis(parent, obj)
73		changeWorkflowState(
74		analysis, "bika_analysis_workflow", "sample_received")
75
76		# Assign the new analysis to this same worksheet, if any.
77		ws = obj.getWorksheet()
78		if ws:
79		ws.addAnalysis(analysis)
80		analysis.reindexObject()
81
82		# retract our dependencies
83		dependencies = obj.getDependencies()
84		for dependency in dependencies:
85		doActionFor(dependency, 'retract')
86
87		# Retract our dependents
88		dependents = obj.getDependents()
89		for dependent in dependents:
90		doActionFor(dependent, 'retract')
91
92		reindex_request(obj)
93
94
95		def after_verify(obj):
96		"""
97		Method triggered after a 'verify' transition for the analysis passed in
98		is performed. Promotes the transition to the Analysis Request and to
99		Worksheet (if the analysis is assigned to any)
100		This function is called automatically by
101		bika.lims.workfow.AfterTransitionEventHandler
102		"""
103
104		# If the analysis has dependencies, transition them
105		for dependency in obj.getDependencies():
106		doActionFor(dependency, 'verify')
107
108		# Do all the reflex rules process
109		obj._reflex_rule_process('verify')
110
111		# Escalate to Analysis Request. Note that the guard for verify transition
112		# from Analysis Request will check if the AR can be transitioned, so there
113		# is no need to check here if all analyses within the AR have been
114		# transitioned already.
115		ar = obj.getRequest()
116		doActionFor(ar, 'verify')
117
118		# Ecalate to Worksheet. Note that the guard for verify transition from
119		# Worksheet will check if the Worksheet can be transitioned, so there is no
120		# need to check here if all analyses within the WS have been transitioned
121		# already
122		ws = obj.getWorksheet()
123		if ws:
124		doActionFor(ws, 'verify')
125		reindex_request(obj)
126
127
128		def after_assign(obj):
129		"""Function triggered after an 'assign' transition for the analysis passed
130		in is performed."""
131		# Reindex the entire request to update the FieldIndex `assigned_state`
132		reindex_request(obj, idxs=['assigned_state',])
133
134
135		def after_unassign(obj):
136		"""Function triggered after an 'unassign' transition for the analysis passed
137		in is performed."""
138		# Reindex the entire request to update the FieldIndex `assigned_state`
139		reindex_request(obj, idxs=['assigned_state',])
140
141
142		def after_cancel(obj):
143		if skip(obj, "cancel"):
144		return
145		workflow = getToolByName(obj, "portal_workflow")
146		# If it is assigned to a worksheet, unassign it.
147		state = workflow.getInfoFor(obj, 'worksheetanalysis_review_state')
148		if state == 'assigned':
149		ws = obj.getWorksheet()
150		skip(obj, "cancel", unskip=True)
151		ws.removeAnalysis(obj)
152		obj.reindexObject()
153		reindex_request(obj)
154
155
156		def after_reject(obj):
157		if skip(obj, "reject"):
158		return
159		workflow = getToolByName(obj, "portal_workflow")
160		# If it is assigned to a worksheet, unassign it.
161		state = workflow.getInfoFor(obj, 'worksheetanalysis_review_state')
162		if state == 'assigned':
163		ws = obj.getWorksheet()
164		ws.removeAnalysis(obj)
165		obj.reindexObject()
166		reindex_request(obj)
167
168
169		def after_attach(obj):
170		if skip(obj, "attach"):
171		return
172		workflow = getToolByName(obj, "portal_workflow")
173		# If all analyses in this AR have been attached escalate the action
174		# to the parent AR
175		ar = obj.aq_parent
176		state = workflow.getInfoFor(ar, "review_state")
177		if state == "attachment_due" and not skip(ar, "attach", peek=True):
178		can_attach = True
179		for a in ar.getAnalyses():
180		if a.review_state in ("to_be_sampled", "to_be_preserved",
181		"sample_due", "sample_received",
182		"attachment_due"):
183		can_attach = False
184		break
185		if can_attach:
186		workflow.doActionFor(ar, "attach")
187		# If assigned to a worksheet and all analyses on the worksheet have
188		# been attached, then attach the worksheet.
189		ws = obj.getBackReferences('WorksheetAnalysis')
190		if ws:
191		ws_state = workflow.getInfoFor(ws, "review_state")
192	View Code Duplication	if ws_state == "attachment_due" \
		0 ignored issues – show Duplication introduced 2018-07-25 08:58 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
193		and not skip(ws, "attach", peek=True):
194		can_attach = True
195		for a in ws.getAnalyses():
196		state = workflow.getInfoFor(a, "review_state")
197		if state in ("to_be_sampled", "to_be_preserved",
198		"sample_due", "sample_received",
199		"attachment_due", "assigned"):
200		can_attach = False
201		break
202		if can_attach:
203		workflow.doActionFor(ws, "attach")
204		obj.reindexObject()
205		reindex_request(obj)
206
207
208		# TODO Workflow - Analysis - revisit reindexing of ancestors
209		def reindex_request(analysis, idxs=None):
210		"""Reindex the Analysis Request the analysis belongs to, as well as the
211		ancestors recursively
212		"""
213		if not IRequestAnalysis.providedBy(analysis) or \
214		IDuplicateAnalysis.providedBy(analysis):
215		# Analysis not directly bound to an Analysis Request. Do nothing
216		return
217
218		n_idxs = ['assigned_state', 'isRootAncestor', 'getDueDate']
219		n_idxs = idxs and list(set(idxs + n_idxs)) or n_idxs
220		analysis_request = analysis.getRequest()
221		analysis_request.reindexObject(idxs=n_idxs)
222		ancestors = analysis_request.getAncestors(all_ancestors=True)
223		for ancestor in ancestors:
224		ancestor.reindexObject(idxs=n_idxs)
225

senaite / senaite.core

Push — master ( 42e1f6...ef1ceb )

bika.lims.workflow.analysis.events.after_retract() A

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like