PolicyController.get_one() - Code Metrics - Inspection of "Merge pull request #3364 from StackStorm/rbac_for_..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Test Setup Failed

Push — master ( 561579...99c26c )

unknown

created 2017-05-16 10:00 UTC

PolicyController.get_one() A

↳ Parent: PolicyController

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	4
rs	10

# 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, **raw_filters):
        return self._get_all(sort=sort,
                             offset=offset,
                             limit=limit,
                             raw_filters=raw_filters)

    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):

        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)

        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, **raw_filters):
        return self._get_all(sort=sort,
                             offset=offset,
                             limit=limit,
                             raw_filters=raw_filters)

    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, **raw_filters):
53			return self._get_all(sort=sort,
54			offset=offset,
55			limit=limit,
56			raw_filters=raw_filters)
57
58			def _get_one(self, ref_or_id, requester_user):
59			instance = self._get_by_ref_or_id(ref_or_id=ref_or_id)
60
61			permission_type = PermissionType.POLICY_TYPE_VIEW
62			rbac_utils.assert_user_has_resource_db_permission(user_db=requester_user,
63			resource_db=instance,
64			permission_type=permission_type)
65
66			result = self.model.from_model(instance)
67
68			if result and self.include_reference:
69			resource_type = getattr(result, 'resource_type', None)
70			name = getattr(result, 'name', None)
71			result.ref = PolicyTypeReference(resource_type=resource_type, name=name).ref
72
73			return result
74
75			def _get_all(self, exclude_fields=None, sort=None, offset=0, limit=None, query_options=None,
76			from_model_kwargs=None, raw_filters=None):
77
78			resp = super(PolicyTypeController, self)._get_all(exclude_fields=exclude_fields,
79			sort=sort,
80			offset=offset,
81			limit=limit,
82			query_options=query_options,
83			from_model_kwargs=from_model_kwargs,
84			raw_filters=raw_filters)
85
86			if self.include_reference:
87			result = resp.json
88			for item in result:
89			resource_type = item.get('resource_type', None)
90			name = item.get('name', None)
91			item['ref'] = PolicyTypeReference(resource_type=resource_type, name=name).ref
92			resp.json = result
93
94			return resp
95
96			def _get_by_ref_or_id(self, ref_or_id):
97			if PolicyTypeReference.is_reference(ref_or_id):
98			resource_db = self._get_by_ref(resource_ref=ref_or_id)
99			else:
100			resource_db = self._get_by_id(resource_id=ref_or_id)
101
102			if not resource_db:
103			msg = 'PolicyType with a reference of id "%s" not found.' % (ref_or_id)
104			raise StackStormDBObjectNotFoundError(msg)
105
106			return resource_db
107
108			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...
109			try:
110			resource_db = self.access.get_by_id(resource_id)
111			except Exception:
112			resource_db = None
113
114			return resource_db
115
116			def _get_by_ref(self, resource_ref):
117			try:
118			ref = PolicyTypeReference.from_string_reference(ref=resource_ref)
119			except Exception:
120			return None
121
122			resource_db = self.access.query(name=ref.name, resource_type=ref.resource_type).first()
123			return resource_db
124
125
126			class PolicyController(resource.ContentPackResourceController):
127			model = PolicyAPI
128			access = Policy
129
130			supported_filters = {
131			'pack': 'pack',
132			'resource_ref': 'resource_ref',
133			'policy_type': 'policy_type'
134			}
135
136			query_options = {
137			'sort': ['pack', 'name']
138			}
139
140			def get_all(self, sort=None, offset=0, limit=None, **raw_filters):
141			return self._get_all(sort=sort,
142			offset=offset,
143			limit=limit,
144			raw_filters=raw_filters)
145
146			def get_one(self, ref_or_id, requester_user):
147			permission_type = PermissionType.POLICY_VIEW
148			return self._get_one(ref_or_id, permission_type=permission_type,
149			requester_user=requester_user)
150
151			def post(self, instance, requester_user):
152			"""
153			Create a new policy.
154			Handles requests:
155			POST /policies/
156			"""
157			permission_type = PermissionType.POLICY_CREATE
158			rbac_utils.assert_user_has_resource_api_permission(user_db=requester_user,
159			resource_api=instance,
160			permission_type=permission_type)
161
162			op = 'POST /policies/'
163
164			db_model = self.model.to_model(instance)
165			LOG.debug('%s verified object: %s', op, db_model)
166
167			db_model = self.access.add_or_update(db_model)
168
169			LOG.debug('%s created object: %s', op, db_model)
170			LOG.audit('Policy created. Policy.id=%s' % (db_model.id), extra={'policy_db': db_model})
171
172			exec_result = self.model.from_model(db_model)
173
174			return Response(json=exec_result, status=http_client.CREATED)
175
176			def put(self, instance, ref_or_id, requester_user):
177			op = 'PUT /policies/%s/' % ref_or_id
178
179			db_model = self._get_by_ref_or_id(ref_or_id=ref_or_id)
180			LOG.debug('%s found object: %s', op, db_model)
181
182			permission_type = PermissionType.POLICY_MODIFY
183			rbac_utils.assert_user_has_resource_db_permission(user_db=requester_user,
184			resource_db=db_model,
185			permission_type=permission_type)
186
187			db_model_id = db_model.id
188
189			try:
190			validate_not_part_of_system_pack(db_model)
191			except ValueValidationException as e:
192			LOG.exception('%s unable to update object from system pack.', op)
193			abort(http_client.BAD_REQUEST, str(e))
194
195			if not getattr(instance, 'pack', None):
196			instance.pack = db_model.pack
197
198			try:
199			db_model = self.model.to_model(instance)
200			db_model.id = db_model_id
201			db_model = self.access.add_or_update(db_model)
202			except (ValidationError, ValueError) as e:
203			LOG.exception('%s unable to update object: %s', op, db_model)
204			abort(http_client.BAD_REQUEST, str(e))
205			return
206
207			LOG.debug('%s updated object: %s', op, db_model)
208			LOG.audit('Policy updated. Policy.id=%s' % (db_model.id), extra={'policy_db': db_model})
209
210			exec_result = self.model.from_model(db_model)
211
212			return Response(json=exec_result, status=http_client.OK)
213
214			def delete(self, ref_or_id, requester_user):
215			"""
216			Delete a policy.
217			Handles requests:
218			POST /policies/1?_method=delete
219			DELETE /policies/1
220			DELETE /policies/mypack.mypolicy
221			"""
222			op = 'DELETE /policies/%s/' % ref_or_id
223
224			db_model = self._get_by_ref_or_id(ref_or_id=ref_or_id)
225			LOG.debug('%s found object: %s', op, db_model)
226
227			permission_type = PermissionType.POLICY_DELETE
228			rbac_utils.assert_user_has_resource_db_permission(user_db=requester_user,
229			resource_db=db_model,
230			permission_type=permission_type)
231
232			try:
233			validate_not_part_of_system_pack(db_model)
234			except ValueValidationException as e:
235			LOG.exception('%s unable to delete object from system pack.', op)
236			abort(http_client.BAD_REQUEST, str(e))
237
238			try:
239			self.access.delete(db_model)
240			except Exception as e:
241			LOG.exception('%s unable to delete object: %s', op, db_model)
242			abort(http_client.INTERNAL_SERVER_ERROR, str(e))
243			return
244
245			LOG.debug('%s deleted object: %s', op, db_model)
246			LOG.audit('Policy deleted. Policy.id=%s' % (db_model.id), extra={'policy_db': db_model})
247
248			# return None
249			return Response(status=http_client.NO_CONTENT)
250
251
252			policy_type_controller = PolicyTypeController()
253			policy_controller = PolicyController()
254

StackStorm / st2

Push — master ( 561579...99c26c )

PolicyController.get_one() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like