bika.lims.workflow.analysis.events.promote_to_dependencies() - Code Metrics - Inspection of "Update worksheet and analysis request when analyse..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 0193a4...9eb485 )

by Ramon

created 2018-11-29 06:12 UTC

promote_to_dependencies() A

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

Complexity

Conditions

Size

Total Lines	6
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# -*- 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 import api
from bika.lims.interfaces import IDuplicateAnalysis
from bika.lims.interfaces.analysis import IRequestAnalysis
from bika.lims.utils.analysis import create_analysis
from bika.lims.workflow import doActionFor, push_reindex_to_actions_pool
from bika.lims.workflow import skip


def after_assign(analysis):
    """Function triggered after an 'assign' transition for the analysis passed
    in is performed.
    """
    reindex_request(analysis)


def after_unassign(analysis):
    """Function triggered after an 'unassign' transition for the analysis passed
    in is performed.
    """
    # Remove from the worksheet
    remove_analysis_from_worksheet(analysis)
    # Reindex the Analysis Request
    reindex_request(analysis)


def after_cancel(analysis):
    """Function triggered after a "cancel" transition is performed. Removes the
    cancelled analysis from the worksheet, if any.
    """
    # Remove from the worksheet
    remove_analysis_from_worksheet(analysis)


def after_reinstate(analysis):
    """Function triggered after a "reinstate" transition is performed.
    """
    pass


def after_submit(analysis):
    """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
    """
    # Promote to analyses this analysis depends on
    promote_to_dependencies(analysis, "submit")

    # TODO: REFLEX TO REMOVE
    # Do all the reflex rules process
    if IRequestAnalysis.providedBy(analysis):
        analysis._reflex_rule_process('submit')

    # Promote transition to worksheet
    ws = analysis.getWorksheet()
    if ws:
        doActionFor(ws, 'submit')

    # Promote transition to Analysis Request
    if IRequestAnalysis.providedBy(analysis):
        doActionFor(analysis.getRequest(), 'submit')
        reindex_request(analysis)


def after_retract(analysis):
    """Function triggered after a 'retract' transition for the analysis passed
    in is performed. The analysis transitions to "retracted" state and a new
    copy of the analysis is created. The copy initial state is "unassigned",
    unless the the retracted analysis was assigned to a worksheet. In such case,
    the copy is transitioned to 'assigned' state too
    """
    # Rename the analysis to make way for it's successor.
    # Support multiple retractions by renaming to *-0, *-1, etc
    parent = analysis.aq_parent
    keyword = analysis.getKeyword()
    analyses = filter(lambda an: an.getKeyword() == keyword,
                      parent.objectValues("Analysis"))

    # Rename the retracted analysis
    # https://docs.plone.org/develop/plone/content/rename.html
    # _verifyObjectPaste permission check must be cancelled
    parent._verifyObjectPaste = str
    retracted_id = '{}-{}'.format(keyword, len(analyses))
    # Make sure all persistent objects have _p_jar attribute
    transaction.savepoint(optimistic=True)
    parent.manage_renameObject(analysis.getId(), retracted_id)
    delattr(parent, '_verifyObjectPaste')

    # Create a copy of the retracted analysis
    analysis_uid = api.get_uid(analysis)
    new_analysis = create_analysis(parent, analysis, RetestOf=analysis_uid)

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

    # Retract our dependents (analyses that depend on this analysis)
    cascade_to_dependents(analysis, "retract")


def after_reject(analysis):
    """Function triggered after the "reject" transition for the analysis passed
    in is performed."""
    worksheet = analysis.getWorksheet()
    if worksheet:
        worksheet.removeAnalysis(analysis)

    # Reject our dependents (analyses that depend on this analysis)
    cascade_to_dependents(analysis, "reject")

    # Try to rollback the Analysis Request (all analyses rejected)
    if IRequestAnalysis.providedBy(analysis):
        doActionFor(analysis.getRequest(), "rollback_to_receive")
        reindex_request(analysis)


def after_verify(analysis):
    """
    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
    """
    # Promote to analyses this analysis depends on
    promote_to_dependencies(analysis, "verify")

    # TODO: REFLEX TO REMOVE
    # Do all the reflex rules process
    if IRequestAnalysis.providedBy(analysis):
        analysis._reflex_rule_process('verify')

    # Promote transition to worksheet
    ws = analysis.getWorksheet()
    if ws:
        doActionFor(ws, 'verify')

    # Promote transition to Analysis Request
    if IRequestAnalysis.providedBy(analysis):
        doActionFor(analysis.getRequest(), 'verify')
        reindex_request(analysis)


def after_publish(analysis):
    """Function triggered after a "publish" transition is performed.
    """
    pass


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 ("unassigned", "assigned", "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.getWorksheet()
    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 ("unassigned", "assigned", "attachment_due"):
                    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', 'getDueDate']
    n_idxs = idxs and list(set(idxs + n_idxs)) or n_idxs
    request = analysis.getRequest()
    ancestors = [request] + request.getAncestors(all_ancestors=True)
    for ancestor in ancestors:
        push_reindex_to_actions_pool(ancestor, n_idxs)


def remove_analysis_from_worksheet(analysis):
    """Removes the analysis passed in from the worksheet, if assigned to any
    """
    worksheet = analysis.getWorksheet()
    if not worksheet:
        return

    # Removal of a routine analysis causes the removal of their duplicates
    if not IDuplicateAnalysis.providedBy(analysis):
        for dup in worksheet.get_duplicates_for(analysis):
            doActionFor(dup, "unassign")

    analyses = filter(lambda an: an != analysis, worksheet.getAnalyses())
    worksheet.setAnalyses(analyses)
    worksheet.purgeLayout()
    if analyses:
        # Maybe this analysis was the only one that was not yet submitted or
        # verified, so try to submit or verify the Worksheet to be aligned
        # with the current states of the analyses it contains.
        doActionFor(worksheet, "submit")
        doActionFor(worksheet, "verify")
    else:
        # We've removed all analyses. Rollback to "open"
        doActionFor(worksheet, "rollback_to_open")

    # Reindex the Worksheet
    idxs = ["getAnalysesUIDs", "getDepartmentUIDs"]
    push_reindex_to_actions_pool(worksheet, idxs=idxs)


def cascade_to_dependents(analysis, transition_id):
    """Cascades the transition to dependent analyses (those that depend on the
    analysis passed in), if any
    """
    for dependent in analysis.getDependents():
        doActionFor(dependent, transition_id)


def promote_to_dependencies(analysis, transition_id):
    """Promotes the transition to the analyses this analysis depends on
    (dependencies), if any
    """
    for dependency in analysis.getDependencies():
        doActionFor(dependency, transition_id)


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
11		from bika.lims import api
12		from bika.lims.interfaces import IDuplicateAnalysis
13		from bika.lims.interfaces.analysis import IRequestAnalysis
14		from bika.lims.utils.analysis import create_analysis
15		from bika.lims.workflow import doActionFor, push_reindex_to_actions_pool
16		from bika.lims.workflow import skip
17
18
19		def after_assign(analysis):
20		"""Function triggered after an 'assign' transition for the analysis passed
21		in is performed.
22		"""
23		reindex_request(analysis)
24
25
26		def after_unassign(analysis):
27		"""Function triggered after an 'unassign' transition for the analysis passed
28		in is performed.
29		"""
30		# Remove from the worksheet
31		remove_analysis_from_worksheet(analysis)
32		# Reindex the Analysis Request
33		reindex_request(analysis)
34
35
36		def after_cancel(analysis):
37		"""Function triggered after a "cancel" transition is performed. Removes the
38		cancelled analysis from the worksheet, if any.
39		"""
40		# Remove from the worksheet
41		remove_analysis_from_worksheet(analysis)
42
43
44		def after_reinstate(analysis):
45		"""Function triggered after a "reinstate" transition is performed.
46		"""
47		pass
48
49
50		def after_submit(analysis):
51		"""Method triggered after a 'submit' transition for the analysis passed in
52		is performed. Promotes the submit transition to the Worksheet to which the
53		analysis belongs to. Note that for the worksheet there is already a guard
54		that assures the transition to the worksheet will only be performed if all
55		analyses within the worksheet have already been transitioned.
56		This function is called automatically by
57		bika.lims.workfow.AfterTransitionEventHandler
58		"""
59		# Promote to analyses this analysis depends on
60		promote_to_dependencies(analysis, "submit")
61
62		# TODO: REFLEX TO REMOVE
63		# Do all the reflex rules process
64		if IRequestAnalysis.providedBy(analysis):
65		analysis._reflex_rule_process('submit')
66
67		# Promote transition to worksheet
68		ws = analysis.getWorksheet()
69		if ws:
70		doActionFor(ws, 'submit')
71
72		# Promote transition to Analysis Request
73		if IRequestAnalysis.providedBy(analysis):
74		doActionFor(analysis.getRequest(), 'submit')
75		reindex_request(analysis)
76
77
78		def after_retract(analysis):
79		"""Function triggered after a 'retract' transition for the analysis passed
80		in is performed. The analysis transitions to "retracted" state and a new
81		copy of the analysis is created. The copy initial state is "unassigned",
82		unless the the retracted analysis was assigned to a worksheet. In such case,
83		the copy is transitioned to 'assigned' state too
84		"""
85		# Rename the analysis to make way for it's successor.
86		# Support multiple retractions by renaming to -0, -1, etc
87		parent = analysis.aq_parent
88		keyword = analysis.getKeyword()
89		analyses = filter(lambda an: an.getKeyword() == keyword,
90		parent.objectValues("Analysis"))
91
92		# Rename the retracted analysis
93		# https://docs.plone.org/develop/plone/content/rename.html
94		# _verifyObjectPaste permission check must be cancelled
95		parent._verifyObjectPaste = str
96		retracted_id = '{}-{}'.format(keyword, len(analyses))
97		# Make sure all persistent objects have _p_jar attribute
98		transaction.savepoint(optimistic=True)
99		parent.manage_renameObject(analysis.getId(), retracted_id)
100		delattr(parent, '_verifyObjectPaste')
101
102		# Create a copy of the retracted analysis
103		analysis_uid = api.get_uid(analysis)
104		new_analysis = create_analysis(parent, analysis, RetestOf=analysis_uid)
105
106		# Assign the new analysis to this same worksheet, if any.
107		worksheet = analysis.getWorksheet()
108		if worksheet:
109		worksheet.addAnalysis(new_analysis)
110
111		# Retract our dependents (analyses that depend on this analysis)
112		cascade_to_dependents(analysis, "retract")
113
114
115		def after_reject(analysis):
116		"""Function triggered after the "reject" transition for the analysis passed
117		in is performed."""
118		worksheet = analysis.getWorksheet()
119		if worksheet:
120		worksheet.removeAnalysis(analysis)
121
122		# Reject our dependents (analyses that depend on this analysis)
123		cascade_to_dependents(analysis, "reject")
124
125		# Try to rollback the Analysis Request (all analyses rejected)
126		if IRequestAnalysis.providedBy(analysis):
127		doActionFor(analysis.getRequest(), "rollback_to_receive")
128		reindex_request(analysis)
129
130
131		def after_verify(analysis):
132		"""
133		Method triggered after a 'verify' transition for the analysis passed in
134		is performed. Promotes the transition to the Analysis Request and to
135		Worksheet (if the analysis is assigned to any)
136		This function is called automatically by
137		bika.lims.workfow.AfterTransitionEventHandler
138		"""
139		# Promote to analyses this analysis depends on
140		promote_to_dependencies(analysis, "verify")
141
142		# TODO: REFLEX TO REMOVE
143		# Do all the reflex rules process
144		if IRequestAnalysis.providedBy(analysis):
145		analysis._reflex_rule_process('verify')
146
147		# Promote transition to worksheet
148		ws = analysis.getWorksheet()
149		if ws:
150		doActionFor(ws, 'verify')
151
152		# Promote transition to Analysis Request
153		if IRequestAnalysis.providedBy(analysis):
154		doActionFor(analysis.getRequest(), 'verify')
155		reindex_request(analysis)
156
157
158		def after_publish(analysis):
159		"""Function triggered after a "publish" transition is performed.
160		"""
161		pass
162
163
164		def after_attach(obj):
165		if skip(obj, "attach"):
166		return
167		workflow = getToolByName(obj, "portal_workflow")
168		# If all analyses in this AR have been attached escalate the action
169		# to the parent AR
170		ar = obj.aq_parent
171		state = workflow.getInfoFor(ar, "review_state")
172		if state == "attachment_due" and not skip(ar, "attach", peek=True):
173		can_attach = True
174		for a in ar.getAnalyses():
175		if a.review_state in ("unassigned", "assigned", "attachment_due"):
176		can_attach = False
177		break
178		if can_attach:
179		workflow.doActionFor(ar, "attach")
180		# If assigned to a worksheet and all analyses on the worksheet have
181		# been attached, then attach the worksheet.
182		ws = obj.getWorksheet()
183		if ws:
184		ws_state = workflow.getInfoFor(ws, "review_state")
185	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...
186		and not skip(ws, "attach", peek=True):
187		can_attach = True
188		for a in ws.getAnalyses():
189		state = workflow.getInfoFor(a, "review_state")
190		if state in ("unassigned", "assigned", "attachment_due"):
191		can_attach = False
192		break
193		if can_attach:
194		workflow.doActionFor(ws, "attach")
195		obj.reindexObject()
196		reindex_request(obj)
197
198
199		# TODO Workflow - Analysis - revisit reindexing of ancestors
200		def reindex_request(analysis, idxs=None):
201		"""Reindex the Analysis Request the analysis belongs to, as well as the
202		ancestors recursively
203		"""
204		if not IRequestAnalysis.providedBy(analysis) or \
205		IDuplicateAnalysis.providedBy(analysis):
206		# Analysis not directly bound to an Analysis Request. Do nothing
207		return
208
209		n_idxs = ['assigned_state', 'getDueDate']
210		n_idxs = idxs and list(set(idxs + n_idxs)) or n_idxs
211		request = analysis.getRequest()
212		ancestors = [request] + request.getAncestors(all_ancestors=True)
213		for ancestor in ancestors:
214		push_reindex_to_actions_pool(ancestor, n_idxs)
215
216
217		def remove_analysis_from_worksheet(analysis):
218		"""Removes the analysis passed in from the worksheet, if assigned to any
219		"""
220		worksheet = analysis.getWorksheet()
221		if not worksheet:
222		return
223
224		# Removal of a routine analysis causes the removal of their duplicates
225		if not IDuplicateAnalysis.providedBy(analysis):
226		for dup in worksheet.get_duplicates_for(analysis):
227		doActionFor(dup, "unassign")
228
229		analyses = filter(lambda an: an != analysis, worksheet.getAnalyses())
230		worksheet.setAnalyses(analyses)
231		worksheet.purgeLayout()
232		if analyses:
233		# Maybe this analysis was the only one that was not yet submitted or
234		# verified, so try to submit or verify the Worksheet to be aligned
235		# with the current states of the analyses it contains.
236		doActionFor(worksheet, "submit")
237		doActionFor(worksheet, "verify")
238		else:
239		# We've removed all analyses. Rollback to "open"
240		doActionFor(worksheet, "rollback_to_open")
241
242		# Reindex the Worksheet
243		idxs = ["getAnalysesUIDs", "getDepartmentUIDs"]
244		push_reindex_to_actions_pool(worksheet, idxs=idxs)
245
246
247		def cascade_to_dependents(analysis, transition_id):
248		"""Cascades the transition to dependent analyses (those that depend on the
249		analysis passed in), if any
250		"""
251		for dependent in analysis.getDependents():
252		doActionFor(dependent, transition_id)
253
254
255		def promote_to_dependencies(analysis, transition_id):
256		"""Promotes the transition to the analyses this analysis depends on
257		(dependencies), if any
258		"""
259		for dependency in analysis.getDependencies():
260		doActionFor(dependency, transition_id)
261

senaite / senaite.core

Push — master ( 0193a4...9eb485 )

promote_to_dependencies() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like