PolicyController.delete() - Code Metrics - Inspection of "Fix wrong information for /api/v1/apikeys/ doc" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#4194)

by Arma

created 2018-06-24 10:28 UTC

PolicyController.delete() A

↳ Parent: PolicyController

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
dl	0
loc	36
rs	9.016
c	0
b	0
f	0

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from mongoengine import ValidationError
from six.moves import http_client

from st2api.controllers import resource
from st2common import log as logging
from st2common.exceptions.apivalidation import ValueValidationException
from st2common.models.api.policy import PolicyTypeAPI, PolicyAPI
from st2common.models.db.policy import PolicyTypeReference
from st2common.persistence.policy import PolicyType, Policy
from st2common.validators.api.misc import validate_not_part_of_system_pack
from st2common.exceptions.db import StackStormDBObjectNotFoundError
from st2common.rbac.types import PermissionType
from st2common.rbac import utils as rbac_utils
from st2common.router import abort
from st2common.router import Response

LOG = logging.getLogger(__name__)


class PolicyTypeController(resource.ResourceController):
    model = PolicyTypeAPI
    access = PolicyType

    supported_filters = {
        'resource_type': 'resource_type'
    }

    query_options = {
        'sort': ['resource_type', 'name']
    }

    include_reference = False

    def get_one(self, ref_or_id, requester_user):
        return self._get_one(ref_or_id, requester_user=requester_user)

    def get_all(self, sort=None, offset=0, limit=None, requester_user=None, **raw_filters):
        return self._get_all(sort=sort,
                             offset=offset,
                             limit=limit,
                             raw_filters=raw_filters,
                             requester_user=requester_user)

    def _get_one(self, ref_or_id, requester_user):
        instance = self._get_by_ref_or_id(ref_or_id=ref_or_id)

        permission_type = PermissionType.POLICY_TYPE_VIEW
        rbac_utils.assert_user_has_resource_db_permission(user_db=requester_user,
                                                          resource_db=instance,
                                                          permission_type=permission_type)

        result = self.model.from_model(instance)

        if result and self.include_reference:
            resource_type = getattr(result, 'resource_type', None)
            name = getattr(result, 'name', None)
            result.ref = PolicyTypeReference(resource_type=resource_type, name=name).ref

        return result

    def _get_all(self, exclude_fields=None, sort=None, offset=0, limit=None, query_options=None,

                 from_model_kwargs=None, raw_filters=None, requester_user=None):

        resp = super(PolicyTypeController, self)._get_all(exclude_fields=exclude_fields,
                                                          sort=sort,
                                                          offset=offset,
                                                          limit=limit,
                                                          query_options=query_options,
                                                          from_model_kwargs=from_model_kwargs,
                                                          raw_filters=raw_filters,
                                                          requester_user=requester_user)

        if self.include_reference:
            result = resp.json
            for item in result:
                resource_type = item.get('resource_type', None)
                name = item.get('name', None)
                item['ref'] = PolicyTypeReference(resource_type=resource_type, name=name).ref
            resp.json = result

        return resp

    def _get_by_ref_or_id(self, ref_or_id):
        if PolicyTypeReference.is_reference(ref_or_id):
            resource_db = self._get_by_ref(resource_ref=ref_or_id)
        else:
            resource_db = self._get_by_id(resource_id=ref_or_id)

        if not resource_db:
            msg = 'PolicyType with a reference of id "%s" not found.' % (ref_or_id)
            raise StackStormDBObjectNotFoundError(msg)

        return resource_db

    def _get_by_id(self, resource_id):

        try:
            resource_db = self.access.get_by_id(resource_id)
        except Exception:
            resource_db = None

        return resource_db

    def _get_by_ref(self, resource_ref):
        try:
            ref = PolicyTypeReference.from_string_reference(ref=resource_ref)
        except Exception:
            return None

        resource_db = self.access.query(name=ref.name, resource_type=ref.resource_type).first()
        return resource_db


class PolicyController(resource.ContentPackResourceController):
    model = PolicyAPI
    access = Policy

    supported_filters = {
        'pack': 'pack',
        'resource_ref': 'resource_ref',
        'policy_type': 'policy_type'
    }

    query_options = {
        'sort': ['pack', 'name']
    }

    def get_all(self, sort=None, offset=0, limit=None, requester_user=None, **raw_filters):
        return self._get_all(sort=sort,
                             offset=offset,
                             limit=limit,
                             raw_filters=raw_filters,
                             requester_user=requester_user)

    def get_one(self, ref_or_id, requester_user):
        permission_type = PermissionType.POLICY_VIEW
        return self._get_one(ref_or_id, permission_type=permission_type,
                             requester_user=requester_user)

    def post(self, instance, requester_user):
        """
            Create a new policy.
            Handles requests:
                POST /policies/
        """
        permission_type = PermissionType.POLICY_CREATE
        rbac_utils.assert_user_has_resource_api_permission(user_db=requester_user,
                                                           resource_api=instance,
                                                           permission_type=permission_type)

        op = 'POST /policies/'

        db_model = self.model.to_model(instance)
        LOG.debug('%s verified object: %s', op, db_model)

        db_model = self.access.add_or_update(db_model)

        LOG.debug('%s created object: %s', op, db_model)
        LOG.audit('Policy created. Policy.id=%s' % (db_model.id), extra={'policy_db': db_model})

        exec_result = self.model.from_model(db_model)

        return Response(json=exec_result, status=http_client.CREATED)

    def put(self, instance, ref_or_id, requester_user):
        op = 'PUT /policies/%s/' % ref_or_id

        db_model = self._get_by_ref_or_id(ref_or_id=ref_or_id)
        LOG.debug('%s found object: %s', op, db_model)

        permission_type = PermissionType.POLICY_MODIFY
        rbac_utils.assert_user_has_resource_db_permission(user_db=requester_user,
                                                          resource_db=db_model,
                                                          permission_type=permission_type)

        db_model_id = db_model.id

        try:
            validate_not_part_of_system_pack(db_model)
        except ValueValidationException as e:
            LOG.exception('%s unable to update object from system pack.', op)
            abort(http_client.BAD_REQUEST, str(e))

        if not getattr(instance, 'pack', None):
            instance.pack = db_model.pack

        try:
            db_model = self.model.to_model(instance)
            db_model.id = db_model_id
            db_model = self.access.add_or_update(db_model)
        except (ValidationError, ValueError) as e:
            LOG.exception('%s unable to update object: %s', op, db_model)
            abort(http_client.BAD_REQUEST, str(e))
            return

        LOG.debug('%s updated object: %s', op, db_model)
        LOG.audit('Policy updated. Policy.id=%s' % (db_model.id), extra={'policy_db': db_model})

        exec_result = self.model.from_model(db_model)

        return Response(json=exec_result, status=http_client.OK)

    def delete(self, ref_or_id, requester_user):
        """
            Delete a policy.
            Handles requests:
                POST /policies/1?_method=delete
                DELETE /policies/1
                DELETE /policies/mypack.mypolicy
        """
        op = 'DELETE /policies/%s/' % ref_or_id

        db_model = self._get_by_ref_or_id(ref_or_id=ref_or_id)
        LOG.debug('%s found object: %s', op, db_model)

        permission_type = PermissionType.POLICY_DELETE
        rbac_utils.assert_user_has_resource_db_permission(user_db=requester_user,
                                                          resource_db=db_model,
                                                          permission_type=permission_type)

        try:
            validate_not_part_of_system_pack(db_model)
        except ValueValidationException as e:
            LOG.exception('%s unable to delete object from system pack.', op)
            abort(http_client.BAD_REQUEST, str(e))

        try:
            self.access.delete(db_model)
        except Exception as e:
            LOG.exception('%s unable to delete object: %s', op, db_model)
            abort(http_client.INTERNAL_SERVER_ERROR, str(e))
            return

        LOG.debug('%s deleted object: %s', op, db_model)
        LOG.audit('Policy deleted. Policy.id=%s' % (db_model.id), extra={'policy_db': db_model})

        # return None
        return Response(status=http_client.NO_CONTENT)


policy_type_controller = PolicyTypeController()
policy_controller = PolicyController()


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			from mongoengine import ValidationError
17			from six.moves import http_client
18
19			from st2api.controllers import resource
20			from st2common import log as logging
21			from st2common.exceptions.apivalidation import ValueValidationException
22			from st2common.models.api.policy import PolicyTypeAPI, PolicyAPI
23			from st2common.models.db.policy import PolicyTypeReference
24			from st2common.persistence.policy import PolicyType, Policy
25			from st2common.validators.api.misc import validate_not_part_of_system_pack
26			from st2common.exceptions.db import StackStormDBObjectNotFoundError
27			from st2common.rbac.types import PermissionType
28			from st2common.rbac import utils as rbac_utils
29			from st2common.router import abort
30			from st2common.router import Response
31
32			LOG = logging.getLogger(__name__)
33
34
35			class PolicyTypeController(resource.ResourceController):
36			model = PolicyTypeAPI
37			access = PolicyType
38
39			supported_filters = {
40			'resource_type': 'resource_type'
41			}
42
43			query_options = {
44			'sort': ['resource_type', 'name']
45			}
46
47			include_reference = False
48
49			def get_one(self, ref_or_id, requester_user):
50			return self._get_one(ref_or_id, requester_user=requester_user)
51
52			def get_all(self, sort=None, offset=0, limit=None, requester_user=None, **raw_filters):
53			return self._get_all(sort=sort,
54			offset=offset,
55			limit=limit,
56			raw_filters=raw_filters,
57			requester_user=requester_user)
58
59			def _get_one(self, ref_or_id, requester_user):
60			instance = self._get_by_ref_or_id(ref_or_id=ref_or_id)
61
62			permission_type = PermissionType.POLICY_TYPE_VIEW
63			rbac_utils.assert_user_has_resource_db_permission(user_db=requester_user,
64			resource_db=instance,
65			permission_type=permission_type)
66
67			result = self.model.from_model(instance)
68
69			if result and self.include_reference:
70			resource_type = getattr(result, 'resource_type', None)
71			name = getattr(result, 'name', None)
72			result.ref = PolicyTypeReference(resource_type=resource_type, name=name).ref
73
74			return result
75
76			def _get_all(self, exclude_fields=None, sort=None, offset=0, limit=None, query_options=None,
			0 ignored issues – show Bug introduced 2018-01-19 13:46 UTC by Report Bug Copy Issue Report Arguments number differs from overridden '_get_all' method Loading history...
77			from_model_kwargs=None, raw_filters=None, requester_user=None):
78
79			resp = super(PolicyTypeController, self)._get_all(exclude_fields=exclude_fields,
80			sort=sort,
81			offset=offset,
82			limit=limit,
83			query_options=query_options,
84			from_model_kwargs=from_model_kwargs,
85			raw_filters=raw_filters,
86			requester_user=requester_user)
87
88			if self.include_reference:
89			result = resp.json
90			for item in result:
91			resource_type = item.get('resource_type', None)
92			name = item.get('name', None)
93			item['ref'] = PolicyTypeReference(resource_type=resource_type, name=name).ref
94			resp.json = result
95
96			return resp
97
98			def _get_by_ref_or_id(self, ref_or_id):
99			if PolicyTypeReference.is_reference(ref_or_id):
100			resource_db = self._get_by_ref(resource_ref=ref_or_id)
101			else:
102			resource_db = self._get_by_id(resource_id=ref_or_id)
103
104			if not resource_db:
105			msg = 'PolicyType with a reference of id "%s" not found.' % (ref_or_id)
106			raise StackStormDBObjectNotFoundError(msg)
107
108			return resource_db
109
110			def _get_by_id(self, resource_id):
			0 ignored issues – show Bug introduced 2015-11-20 23:04 UTC by Report Bug Copy Issue Report Arguments number differs from overridden '_get_by_id' method Loading history...
111			try:
112			resource_db = self.access.get_by_id(resource_id)
113			except Exception:
114			resource_db = None
115
116			return resource_db
117
118			def _get_by_ref(self, resource_ref):
119			try:
120			ref = PolicyTypeReference.from_string_reference(ref=resource_ref)
121			except Exception:
122			return None
123
124			resource_db = self.access.query(name=ref.name, resource_type=ref.resource_type).first()
125			return resource_db
126
127
128			class PolicyController(resource.ContentPackResourceController):
129			model = PolicyAPI
130			access = Policy
131
132			supported_filters = {
133			'pack': 'pack',
134			'resource_ref': 'resource_ref',
135			'policy_type': 'policy_type'
136			}
137
138			query_options = {
139			'sort': ['pack', 'name']
140			}
141
142			def get_all(self, sort=None, offset=0, limit=None, requester_user=None, **raw_filters):
143			return self._get_all(sort=sort,
144			offset=offset,
145			limit=limit,
146			raw_filters=raw_filters,
147			requester_user=requester_user)
148
149			def get_one(self, ref_or_id, requester_user):
150			permission_type = PermissionType.POLICY_VIEW
151			return self._get_one(ref_or_id, permission_type=permission_type,
152			requester_user=requester_user)
153
154			def post(self, instance, requester_user):
155			"""
156			Create a new policy.
157			Handles requests:
158			POST /policies/
159			"""
160			permission_type = PermissionType.POLICY_CREATE
161			rbac_utils.assert_user_has_resource_api_permission(user_db=requester_user,
162			resource_api=instance,
163			permission_type=permission_type)
164
165			op = 'POST /policies/'
166
167			db_model = self.model.to_model(instance)
168			LOG.debug('%s verified object: %s', op, db_model)
169
170			db_model = self.access.add_or_update(db_model)
171
172			LOG.debug('%s created object: %s', op, db_model)
173			LOG.audit('Policy created. Policy.id=%s' % (db_model.id), extra={'policy_db': db_model})
174
175			exec_result = self.model.from_model(db_model)
176
177			return Response(json=exec_result, status=http_client.CREATED)
178
179			def put(self, instance, ref_or_id, requester_user):
180			op = 'PUT /policies/%s/' % ref_or_id
181
182			db_model = self._get_by_ref_or_id(ref_or_id=ref_or_id)
183			LOG.debug('%s found object: %s', op, db_model)
184
185			permission_type = PermissionType.POLICY_MODIFY
186			rbac_utils.assert_user_has_resource_db_permission(user_db=requester_user,
187			resource_db=db_model,
188			permission_type=permission_type)
189
190			db_model_id = db_model.id
191
192			try:
193			validate_not_part_of_system_pack(db_model)
194			except ValueValidationException as e:
195			LOG.exception('%s unable to update object from system pack.', op)
196			abort(http_client.BAD_REQUEST, str(e))
197
198			if not getattr(instance, 'pack', None):
199			instance.pack = db_model.pack
200
201			try:
202			db_model = self.model.to_model(instance)
203			db_model.id = db_model_id
204			db_model = self.access.add_or_update(db_model)
205			except (ValidationError, ValueError) as e:
206			LOG.exception('%s unable to update object: %s', op, db_model)
207			abort(http_client.BAD_REQUEST, str(e))
208			return
209
210			LOG.debug('%s updated object: %s', op, db_model)
211			LOG.audit('Policy updated. Policy.id=%s' % (db_model.id), extra={'policy_db': db_model})
212
213			exec_result = self.model.from_model(db_model)
214
215			return Response(json=exec_result, status=http_client.OK)
216
217			def delete(self, ref_or_id, requester_user):
218			"""
219			Delete a policy.
220			Handles requests:
221			POST /policies/1?_method=delete
222			DELETE /policies/1
223			DELETE /policies/mypack.mypolicy
224			"""
225			op = 'DELETE /policies/%s/' % ref_or_id
226
227			db_model = self._get_by_ref_or_id(ref_or_id=ref_or_id)
228			LOG.debug('%s found object: %s', op, db_model)
229
230			permission_type = PermissionType.POLICY_DELETE
231			rbac_utils.assert_user_has_resource_db_permission(user_db=requester_user,
232			resource_db=db_model,
233			permission_type=permission_type)
234
235			try:
236			validate_not_part_of_system_pack(db_model)
237			except ValueValidationException as e:
238			LOG.exception('%s unable to delete object from system pack.', op)
239			abort(http_client.BAD_REQUEST, str(e))
240
241			try:
242			self.access.delete(db_model)
243			except Exception as e:
244			LOG.exception('%s unable to delete object: %s', op, db_model)
245			abort(http_client.INTERNAL_SERVER_ERROR, str(e))
246			return
247
248			LOG.debug('%s deleted object: %s', op, db_model)
249			LOG.audit('Policy deleted. Policy.id=%s' % (db_model.id), extra={'policy_db': db_model})
250
251			# return None
252			return Response(status=http_client.NO_CONTENT)
253
254
255			policy_type_controller = PolicyTypeController()
256			policy_controller = PolicyController()
257

StackStorm / st2

Pull Request — master (#4194)

PolicyController.delete() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like