bika.lims.workflow.analysis.events.after_publish() - Code Metrics - Inspection of "Do not allow to publish analyses individually (#11..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 1513bd...0193a4 )

by Ramon

created 2018-11-28 21:58 UTC

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

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

Complexity

Conditions

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	2
dl	0
loc	4
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.

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.
    """
    reindex_request(analysis)


def after_cancel(analysis):
    """Function triggered after a "cancel" transition is performed. Removes the
    cancelled analysis from the worksheet, if any.
    """
    worksheet = analysis.getWorksheet()
    if worksheet:
        # Remove the analysis from the worksheet
        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,
            # so try to submit or verify the Worksheet to be consistent 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")


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
    for dependency in analysis.getDependencies():
        doActionFor(dependency, "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)
    for dependent in analysis.getDependents():
        doActionFor(dependent, '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)
    for dependent in analysis.getDependents():
        doActionFor(dependent, "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
    for dependency in analysis.getDependencies():
        doActionFor(dependency, '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)


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

senaite / senaite.core

Push — master ( 1513bd...0193a4 )

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

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like