bika.lims.browser.fields.aranalysesfield.ARAnalysesField._is_frozen() - Code Metrics - Inspection of "Listing: Removed non-conform handling of disabled..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( b17999...915400 )

by Jordi

created 2018-12-06 20:55 UTC

ARAnalysesField._is_frozen() A

↳ Parent: bika.lims.browser.fields.aranalysesfield

Complexity

Conditions

Size

Total Lines	16
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	10
dl	0
loc	16
rs	9.9
c	0
b	0
f	0
cc	4
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 itertools

from AccessControl import ClassSecurityInfo
from AccessControl import Unauthorized
from AccessControl import getSecurityManager
from bika.lims import api
from bika.lims import logger
from bika.lims.catalog import CATALOG_ANALYSIS_LISTING
from bika.lims.interfaces import IAnalysis
from bika.lims.interfaces import IAnalysisService
from bika.lims.interfaces import IARAnalysesField
from bika.lims.permissions import AddAnalysis
from bika.lims.utils.analysis import create_analysis
from Products.Archetypes.public import Field
from Products.Archetypes.public import ObjectField
from Products.Archetypes.Registry import registerField
from Products.Archetypes.utils import shasattr
from Products.CMFCore.utils import getToolByName
from zope.interface import implements

"""Field to manage Analyses on ARs

Please see the assigned doctest at tests/doctests/ARAnalysesField.rst

Run this test from the buildout directory:

    bin/test test_textual_doctests -t ARAnalysesField
"""


class ARAnalysesField(ObjectField):
    """A field that stores Analyses instances
    """
    implements(IARAnalysesField)

    security = ClassSecurityInfo()
    _properties = Field._properties.copy()
    _properties.update({
        "type": "analyses",
        "default": None,
    })

    security.declarePrivate('get')

    def get(self, instance, **kwargs):
        """Returns a list of Analyses assigned to this AR

        Return a list of catalog brains unless `full_objects=True` is passed.
        Other keyword arguments are passed to bika_analysis_catalog

        :param instance: Analysis Request object
        :param kwargs: Keyword arguments to inject in the search query
        :returns: A list of Analysis Objects/Catalog Brains
        """
        catalog = getToolByName(instance, CATALOG_ANALYSIS_LISTING)
        query = dict(
            [(k, v) for k, v in kwargs.items() if k in catalog.indexes()])
        query["portal_type"] = "Analysis"
        query["getRequestUID"] = api.get_uid(instance)
        analyses = catalog(query)
        if not kwargs.get("full_objects", False):
            return analyses

        return map(api.get_object, analyses)

    security.declarePrivate('set')

    def set(self, instance, items, prices=None, specs=None, **kwargs):
        """Set/Assign Analyses to this AR

        :param items: List of Analysis objects/brains, AnalysisService
                      objects/brains and/or Analysis Service uids
        :type items: list
        :param prices: Mapping of AnalysisService UID -> price
        :type prices: dict
        :param specs: List of AnalysisService UID -> Result Range mappings
        :type specs: list
        :returns: list of new assigned Analyses
        """

        # This setter returns a list of new set Analyses
        new_analyses = []

        # Current assigned analyses
        analyses = instance.objectValues("Analysis")

        # Analyses which are in a non-open state must be retained
        non_open_analyses = filter(lambda an: not an.isOpen(), analyses)

        # Prevent removing all analyses
        #
        # N.B.: Non-open analyses are rendered disabled in the HTML form.
        #       Therefore, their UIDs are not included in the submitted UIDs.
        if not items and not non_open_analyses:
            logger.warn("Not allowed to remove all Analyses from AR.")
            return new_analyses

        # Bail out if the items is not a list type
        if not isinstance(items, (list, tuple)):
            raise TypeError(
                "Items parameter must be a tuple or list, got '{}'".format(
                    type(items)))

        # Bail out if the AR is inactive
        if not api.is_active(instance):
            raise Unauthorized("Inactive ARs can not be modified"
                               .format(AddAnalysis))

        # Bail out if the user has not the right permission
        sm = getSecurityManager()
        if not sm.checkPermission(AddAnalysis, instance):
            raise Unauthorized("You do not have the '{}' permission"
                               .format(AddAnalysis))

        # Convert the items to a valid list of AnalysisServices
        services = filter(None, map(self._to_service, items))

        # Calculate dependencies
        # FIXME Infinite recursion error possible here, if the formula includes
        #       the Keyword of the Service that includes the Calculation
        dependencies = map(lambda s: s.getServiceDependencies(), services)
        dependencies = list(itertools.chain.from_iterable(dependencies))

        # Merge dependencies and services
        services = set(services + dependencies)

        # Modify existing AR specs with new form values of selected analyses.
        self._update_specs(instance, specs)

        # CREATE/MODIFY ANALYSES

        for service in services:
            keyword = service.getKeyword()

            # Create the Analysis if it doesn't exist
            if shasattr(instance, keyword):
                analysis = instance._getOb(keyword)
            else:
                analysis = create_analysis(instance, service)
                new_analyses.append(analysis)

            # Set the price of the Analysis
            self._update_price(analysis, service, prices)

        # DELETE ANALYSES

        # Service UIDs
        service_uids = map(api.get_uid, services)

        # Analyses IDs to delete
        delete_ids = []

        # Assigned Attachments
        assigned_attachments = []

        for analysis in analyses:
            service_uid = analysis.getServiceUID()

            # Skip if the Service is selected
            if service_uid in service_uids:
                continue

            # Skip non-open Analyses
            if analysis in non_open_analyses:
                continue

            # Remember assigned attachments
            # https://github.com/senaite/senaite.core/issues/1025
            assigned_attachments.extend(analysis.getAttachment())
            analysis.setAttachment([])

            # If it is assigned to a worksheet, unassign it before deletion.
            worksheet = analysis.getWorksheet()
            if worksheet:
                worksheet.removeAnalysis(analysis)

            # Unset the partition reference
            part = analysis.getSamplePartition()
            if part:
                # From this partition, remove the reference to the current
                # analysis that is going to be removed to prevent inconsistent
                # states (Sample Partitions referencing to Analyses that do not
                # exist anymore
                an_uid = api.get_uid(analysis)
                part_ans = part.getAnalyses() or []
                part_ans = filter(
                    lambda an: api.get_uid(an) != an_uid, part_ans)

                part.setAnalyses(part_ans)
            # Unset the Analysis-to-Partition reference
            analysis.setSamplePartition(None)
            delete_ids.append(analysis.getId())

        if delete_ids:
            # Note: subscriber might promote the AR
            instance.manage_delObjects(ids=delete_ids)

        # Remove orphaned attachments
        for attachment in assigned_attachments:
            # only delete attachments which are no further linked
            if not attachment.getLinkedAnalyses():
                logger.info(
                    "Deleting attachment: {}".format(attachment.getId()))
                attachment_id = api.get_id(attachment)
                api.get_parent(attachment).manage_delObjects(attachment_id)

        return new_analyses

    def _get_services(self, full_objects=False):
        """Fetch and return analysis service objects
        """
        bsc = api.get_tool("bika_setup_catalog")
        brains = bsc(portal_type="AnalysisService")
        if full_objects:
            return map(api.get_object, brains)
        return brains

    def _to_service(self, thing):
        """Convert to Analysis Service

        :param thing: UID/Catalog Brain/Object/Something
        :returns: Analysis Service object or None
        """

        # Convert UIDs to objects
        if api.is_uid(thing):
            thing = api.get_object_by_uid(thing, None)

        # Bail out if the thing is not a valid object
        if not api.is_object(thing):
            logger.warn("'{}' is not a valid object!".format(repr(thing)))
            return None

        # Ensure we have an object here and not a brain
        obj = api.get_object(thing)

        if IAnalysisService.providedBy(obj):
            return obj

        if IAnalysis.providedBy(obj):
            return obj.getAnalysisService()

        # An object, but neither an Analysis nor AnalysisService?
        # This should never happen.
        portal_type = api.get_portal_type(obj)
        logger.error("ARAnalysesField doesn't accept objects from {} type. "
                     "The object will be dismissed.".format(portal_type))
        return None

    def _update_price(self, analysis, service, prices):
        """Update the Price of the Analysis

        :param analysis: Analysis Object
        :param service: Analysis Service Object
        :param prices: Price mapping
        """
        prices = prices or {}
        price = prices.get(service.UID(), service.getPrice())
        analysis.setPrice(price)

    def _update_specs(self, instance, specs):
        """Update AR specifications

        :param instance: Analysis Request
        :param specs: List of Specification Records
        """

        if specs is None:
            return

        # N.B. we copy the records here, otherwise the spec will be written to
        #      the attached specification of this AR
        rr = {item["keyword"]: item.copy()
              for item in instance.getResultsRange()}
        for spec in specs:
            keyword = spec.get("keyword")
            if keyword in rr:
                # overwrite the instance specification only, if the specific
                # analysis spec has min/max values set
                if all([spec.get("min"), spec.get("max")]):
                    rr[keyword].update(spec)
                else:
                    rr[keyword] = spec
            else:
                rr[keyword] = spec
        return instance.setResultsRange(rr.values())


registerField(ARAnalysesField,
              title="Analyses",
              description="Manages Analyses of ARs")


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 itertools
9
10			from AccessControl import ClassSecurityInfo
11			from AccessControl import Unauthorized
12			from AccessControl import getSecurityManager
13			from bika.lims import api
14			from bika.lims import logger
15			from bika.lims.catalog import CATALOG_ANALYSIS_LISTING
16			from bika.lims.interfaces import IAnalysis
17			from bika.lims.interfaces import IAnalysisService
18			from bika.lims.interfaces import IARAnalysesField
19			from bika.lims.permissions import AddAnalysis
20			from bika.lims.utils.analysis import create_analysis
21			from Products.Archetypes.public import Field
22			from Products.Archetypes.public import ObjectField
23			from Products.Archetypes.Registry import registerField
24			from Products.Archetypes.utils import shasattr
25			from Products.CMFCore.utils import getToolByName
26			from zope.interface import implements
27
28			"""Field to manage Analyses on ARs
29
30			Please see the assigned doctest at tests/doctests/ARAnalysesField.rst
31
32			Run this test from the buildout directory:
33
34			bin/test test_textual_doctests -t ARAnalysesField
35			"""
36
37
38			class ARAnalysesField(ObjectField):
39			"""A field that stores Analyses instances
40			"""
41			implements(IARAnalysesField)
42
43			security = ClassSecurityInfo()
44			_properties = Field._properties.copy()
45			_properties.update({
46			"type": "analyses",
47			"default": None,
48			})
49
50			security.declarePrivate('get')
51
52			def get(self, instance, **kwargs):
53			"""Returns a list of Analyses assigned to this AR
54
55			Return a list of catalog brains unless `full_objects=True` is passed.
56			Other keyword arguments are passed to bika_analysis_catalog
57
58			:param instance: Analysis Request object
59			:param kwargs: Keyword arguments to inject in the search query
60			:returns: A list of Analysis Objects/Catalog Brains
61			"""
62			catalog = getToolByName(instance, CATALOG_ANALYSIS_LISTING)
63			query = dict(
64			[(k, v) for k, v in kwargs.items() if k in catalog.indexes()])
65			query["portal_type"] = "Analysis"
66			query["getRequestUID"] = api.get_uid(instance)
67			analyses = catalog(query)
68			if not kwargs.get("full_objects", False):
69			return analyses
70
71			return map(api.get_object, analyses)
72
73			security.declarePrivate('set')
74
75			def set(self, instance, items, prices=None, specs=None, **kwargs):
76			"""Set/Assign Analyses to this AR
77
78			:param items: List of Analysis objects/brains, AnalysisService
79			objects/brains and/or Analysis Service uids
80			:type items: list
81			:param prices: Mapping of AnalysisService UID -> price
82			:type prices: dict
83			:param specs: List of AnalysisService UID -> Result Range mappings
84			:type specs: list
85			:returns: list of new assigned Analyses
86			"""
87
88			# This setter returns a list of new set Analyses
89			new_analyses = []
90
91			# Current assigned analyses
92			analyses = instance.objectValues("Analysis")
93
94			# Analyses which are in a non-open state must be retained
95			non_open_analyses = filter(lambda an: not an.isOpen(), analyses)
96
97			# Prevent removing all analyses
98			#
99			# N.B.: Non-open analyses are rendered disabled in the HTML form.
100			# Therefore, their UIDs are not included in the submitted UIDs.
101			if not items and not non_open_analyses:
102			logger.warn("Not allowed to remove all Analyses from AR.")
103			return new_analyses
104
105			# Bail out if the items is not a list type
106			if not isinstance(items, (list, tuple)):
107			raise TypeError(
108			"Items parameter must be a tuple or list, got '{}'".format(
109			type(items)))
110
111			# Bail out if the AR is inactive
112			if not api.is_active(instance):
113			raise Unauthorized("Inactive ARs can not be modified"
114			.format(AddAnalysis))
115
116			# Bail out if the user has not the right permission
117			sm = getSecurityManager()
118			if not sm.checkPermission(AddAnalysis, instance):
119			raise Unauthorized("You do not have the '{}' permission"
120			.format(AddAnalysis))
121
122			# Convert the items to a valid list of AnalysisServices
123			services = filter(None, map(self._to_service, items))
124
125			# Calculate dependencies
126			# FIXME Infinite recursion error possible here, if the formula includes
127			# the Keyword of the Service that includes the Calculation
128			dependencies = map(lambda s: s.getServiceDependencies(), services)
129			dependencies = list(itertools.chain.from_iterable(dependencies))
130
131			# Merge dependencies and services
132			services = set(services + dependencies)
133
134			# Modify existing AR specs with new form values of selected analyses.
135			self._update_specs(instance, specs)
136
137			# CREATE/MODIFY ANALYSES
138
139			for service in services:
140			keyword = service.getKeyword()
141
142			# Create the Analysis if it doesn't exist
143			if shasattr(instance, keyword):
144			analysis = instance._getOb(keyword)
145			else:
146			analysis = create_analysis(instance, service)
147			new_analyses.append(analysis)
148
149			# Set the price of the Analysis
150			self._update_price(analysis, service, prices)
151
152			# DELETE ANALYSES
153
154			# Service UIDs
155			service_uids = map(api.get_uid, services)
156
157			# Analyses IDs to delete
158			delete_ids = []
159
160			# Assigned Attachments
161			assigned_attachments = []
162
163			for analysis in analyses:
164			service_uid = analysis.getServiceUID()
165
166			# Skip if the Service is selected
167			if service_uid in service_uids:
168			continue
169
170			# Skip non-open Analyses
171			if analysis in non_open_analyses:
172			continue
173
174			# Remember assigned attachments
175			# https://github.com/senaite/senaite.core/issues/1025
176			assigned_attachments.extend(analysis.getAttachment())
177			analysis.setAttachment([])
178
179			# If it is assigned to a worksheet, unassign it before deletion.
180			worksheet = analysis.getWorksheet()
181			if worksheet:
182			worksheet.removeAnalysis(analysis)
183
184			# Unset the partition reference
185			part = analysis.getSamplePartition()
186			if part:
187			# From this partition, remove the reference to the current
188			# analysis that is going to be removed to prevent inconsistent
189			# states (Sample Partitions referencing to Analyses that do not
190			# exist anymore
191			an_uid = api.get_uid(analysis)
192			part_ans = part.getAnalyses() or []
193			part_ans = filter(
194			lambda an: api.get_uid(an) != an_uid, part_ans)
			0 ignored issues – show introduced 2018-12-06 14:54 UTC by Report Bug Copy Issue Report The variable `an_uid` does not seem to be defined for all execution paths. Loading history...
195			part.setAnalyses(part_ans)
196			# Unset the Analysis-to-Partition reference
197			analysis.setSamplePartition(None)
198			delete_ids.append(analysis.getId())
199
200			if delete_ids:
201			# Note: subscriber might promote the AR
202			instance.manage_delObjects(ids=delete_ids)
203
204			# Remove orphaned attachments
205			for attachment in assigned_attachments:
206			# only delete attachments which are no further linked
207			if not attachment.getLinkedAnalyses():
208			logger.info(
209			"Deleting attachment: {}".format(attachment.getId()))
210			attachment_id = api.get_id(attachment)
211			api.get_parent(attachment).manage_delObjects(attachment_id)
212
213			return new_analyses
214
215			def _get_services(self, full_objects=False):
216			"""Fetch and return analysis service objects
217			"""
218			bsc = api.get_tool("bika_setup_catalog")
219			brains = bsc(portal_type="AnalysisService")
220			if full_objects:
221			return map(api.get_object, brains)
222			return brains
223
224			def _to_service(self, thing):
225			"""Convert to Analysis Service
226
227			:param thing: UID/Catalog Brain/Object/Something
228			:returns: Analysis Service object or None
229			"""
230
231			# Convert UIDs to objects
232			if api.is_uid(thing):
233			thing = api.get_object_by_uid(thing, None)
234
235			# Bail out if the thing is not a valid object
236			if not api.is_object(thing):
237			logger.warn("'{}' is not a valid object!".format(repr(thing)))
238			return None
239
240			# Ensure we have an object here and not a brain
241			obj = api.get_object(thing)
242
243			if IAnalysisService.providedBy(obj):
244			return obj
245
246			if IAnalysis.providedBy(obj):
247			return obj.getAnalysisService()
248
249			# An object, but neither an Analysis nor AnalysisService?
250			# This should never happen.
251			portal_type = api.get_portal_type(obj)
252			logger.error("ARAnalysesField doesn't accept objects from {} type. "
253			"The object will be dismissed.".format(portal_type))
254			return None
255
256			def _update_price(self, analysis, service, prices):
257			"""Update the Price of the Analysis
258
259			:param analysis: Analysis Object
260			:param service: Analysis Service Object
261			:param prices: Price mapping
262			"""
263			prices = prices or {}
264			price = prices.get(service.UID(), service.getPrice())
265			analysis.setPrice(price)
266
267			def _update_specs(self, instance, specs):
268			"""Update AR specifications
269
270			:param instance: Analysis Request
271			:param specs: List of Specification Records
272			"""
273
274			if specs is None:
275			return
276
277			# N.B. we copy the records here, otherwise the spec will be written to
278			# the attached specification of this AR
279			rr = {item["keyword"]: item.copy()
280			for item in instance.getResultsRange()}
281			for spec in specs:
282			keyword = spec.get("keyword")
283			if keyword in rr:
284			# overwrite the instance specification only, if the specific
285			# analysis spec has min/max values set
286			if all([spec.get("min"), spec.get("max")]):
287			rr[keyword].update(spec)
288			else:
289			rr[keyword] = spec
290			else:
291			rr[keyword] = spec
292			return instance.setResultsRange(rr.values())
293
294
295			registerField(ARAnalysesField,
296			title="Analyses",
297			description="Manages Analyses of ARs")
298

senaite / senaite.core

Push — master ( b17999...915400 )

ARAnalysesField._is_frozen() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like