validate_trigger_parameters() - Code Metrics - Inspection of "Allow Jinja variables in trigger with parameters" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#3804)

by Lakshmi

created 2017-10-23 15:30 UTC

validate_trigger_parameters() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	9
c	1
b	0
f	0
dl	0
loc	54
rs	5.4234

How to fix Long Method

# 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.

import six

from oslo_config import cfg
from apscheduler.triggers.cron import CronTrigger

from st2common import log as logging
from st2common.exceptions.apivalidation import ValueValidationException
from st2common.exceptions.triggers import TriggerParametersValidationException
from st2common.constants.triggers import SYSTEM_TRIGGER_TYPES
from st2common.constants.triggers import CRON_TIMER_TRIGGER_REF
from st2common.util import schema as util_schema
import st2common.operators as criteria_operators
from st2common.services import triggers

__all__ = [
    'validate_criteria',

    'validate_trigger_parameters',
    'validate_trigger_payload'
]


LOG = logging.getLogger(__name__)

allowed_operators = criteria_operators.get_allowed_operators()


def validate_criteria(criteria):
    if not isinstance(criteria, dict):
        raise ValueValidationException('Criteria should be a dict.')

    for key, value in six.iteritems(criteria):
        operator = value.get('type', None)
        if operator is None:
            raise ValueValidationException('Operator not specified for field: ' + key)
        if operator not in allowed_operators:
            raise ValueValidationException('For field: ' + key + ', operator ' + operator +
                                           ' not in list of allowed operators: ' +
                                           str(allowed_operators.keys()))
        pattern = value.get('pattern', None)
        if pattern is None:
            raise ValueValidationException('For field: ' + key + ', no pattern specified ' +
                                           'for operator ' + operator)


def validate_trigger_parameters(trigger_type_ref, parameters):
    """
    This function validates parameters for system and user-defined triggers.

    :param trigger_type_ref: Reference of a trigger type.
    :type trigger_type_ref: ``str``

    :param parameters: Trigger parameters.
    :type parameters: ``dict``

    :return: Cleaned parameters on success, None if validation is not performed.
    """
    if not trigger_type_ref:
        return None

    is_system_trigger = trigger_type_ref in SYSTEM_TRIGGER_TYPES
    if is_system_trigger:
        # System trigger
        parameters_schema = SYSTEM_TRIGGER_TYPES[trigger_type_ref]['parameters_schema']
    else:
        trigger_type_db = triggers.get_trigger_type_db(trigger_type_ref)
        if not trigger_type_db:
            # Trigger doesn't exist in the database
            return None

        parameters_schema = getattr(trigger_type_db, 'parameters_schema', {})
        if not parameters_schema:
            # Parameters schema not defined for the this trigger
            return None

    # We only validate non-system triggers if config option is set (enabled)
    if not is_system_trigger and not cfg.CONF.system.validate_trigger_parameters:
        LOG.debug('Got non-system trigger "%s", but trigger parameter validation for non-system'

                  'triggers is disabled, skipping validation.' % (trigger_type_ref))
        return None

    cleaned = util_schema.validate(instance=parameters, schema=parameters_schema,
                                   cls=util_schema.CustomValidator, use_default=True,
                                   allow_default_none=True)

    # Additional validation for CronTimer trigger
    # TODO: If we need to add more checks like this we should consider abstracting this out.
    if trigger_type_ref == CRON_TIMER_TRIGGER_REF:
        # Validate that the user provided parameters are valid. This is required since JSON schema
        # allows arbitrary strings, but not any arbitrary string is a valid CronTrigger argument
        # Note: Constructor throws ValueError on invalid parameters
        try:
            CronTrigger(**parameters)
        except:
            msg = ('Cron trigger parameters do not match expected format.')
            LOG.exception(msg)
            raise TriggerParametersValidationException(msg)

    return cleaned


def validate_trigger_payload(trigger_type_ref, payload):
    """
    This function validates trigger payload parameters for system and user-defined triggers.

    :param trigger_type_ref: Reference of a trigger type.
    :type trigger_type_ref: ``str``

    :param payload: Trigger payload.
    :type payload: ``dict``

    :return: Cleaned payload on success, None if validation is not performed.
    """
    if not trigger_type_ref:
        return None

    is_system_trigger = trigger_type_ref in SYSTEM_TRIGGER_TYPES
    if is_system_trigger:
        # System trigger
        payload_schema = SYSTEM_TRIGGER_TYPES[trigger_type_ref]['payload_schema']
    else:
        trigger_type_db = triggers.get_trigger_type_db(trigger_type_ref)
        if not trigger_type_db:
            # Trigger doesn't exist in the database
            return None

        payload_schema = getattr(trigger_type_db, 'payload_schema', {})
        if not payload_schema:
            # Payload schema not defined for the this trigger
            return None

    # We only validate non-system triggers if config option is set (enabled)
    if not is_system_trigger and not cfg.CONF.system.validate_trigger_payload:
        LOG.debug('Got non-system trigger "%s", but trigger payload validation for non-system'

                  'triggers is disabled, skipping validation.' % (trigger_type_ref))
        return None

    cleaned = util_schema.validate(instance=payload, schema=payload_schema,
                                   cls=util_schema.CustomValidator, use_default=True,
                                   allow_default_none=True)

    return cleaned


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			import six
17
18			from oslo_config import cfg
19			from apscheduler.triggers.cron import CronTrigger
20
21			from st2common import log as logging
22			from st2common.exceptions.apivalidation import ValueValidationException
23			from st2common.exceptions.triggers import TriggerParametersValidationException
24			from st2common.constants.triggers import SYSTEM_TRIGGER_TYPES
25			from st2common.constants.triggers import CRON_TIMER_TRIGGER_REF
26			from st2common.util import schema as util_schema
27			import st2common.operators as criteria_operators
28			from st2common.services import triggers
29
30			__all__ = [
31			'validate_criteria',
32
33			'validate_trigger_parameters',
34			'validate_trigger_payload'
35			]
36
37
38			LOG = logging.getLogger(__name__)
39
40			allowed_operators = criteria_operators.get_allowed_operators()
41
42
43			def validate_criteria(criteria):
44			if not isinstance(criteria, dict):
45			raise ValueValidationException('Criteria should be a dict.')
46
47			for key, value in six.iteritems(criteria):
48			operator = value.get('type', None)
49			if operator is None:
50			raise ValueValidationException('Operator not specified for field: ' + key)
51			if operator not in allowed_operators:
52			raise ValueValidationException('For field: ' + key + ', operator ' + operator +
53			' not in list of allowed operators: ' +
54			str(allowed_operators.keys()))
55			pattern = value.get('pattern', None)
56			if pattern is None:
57			raise ValueValidationException('For field: ' + key + ', no pattern specified ' +
58			'for operator ' + operator)
59
60
61			def validate_trigger_parameters(trigger_type_ref, parameters):
62			"""
63			This function validates parameters for system and user-defined triggers.
64
65			:param trigger_type_ref: Reference of a trigger type.
66			:type trigger_type_ref: ``str``
67
68			:param parameters: Trigger parameters.
69			:type parameters: ``dict``
70
71			:return: Cleaned parameters on success, None if validation is not performed.
72			"""
73			if not trigger_type_ref:
74			return None
75
76			is_system_trigger = trigger_type_ref in SYSTEM_TRIGGER_TYPES
77			if is_system_trigger:
78			# System trigger
79			parameters_schema = SYSTEM_TRIGGER_TYPES[trigger_type_ref]['parameters_schema']
80			else:
81			trigger_type_db = triggers.get_trigger_type_db(trigger_type_ref)
82			if not trigger_type_db:
83			# Trigger doesn't exist in the database
84			return None
85
86			parameters_schema = getattr(trigger_type_db, 'parameters_schema', {})
87			if not parameters_schema:
88			# Parameters schema not defined for the this trigger
89			return None
90
91			# We only validate non-system triggers if config option is set (enabled)
92			if not is_system_trigger and not cfg.CONF.system.validate_trigger_parameters:
93			LOG.debug('Got non-system trigger "%s", but trigger parameter validation for non-system'
			0 ignored issues – show Coding Style Best Practice introduced 2016-12-27 15:05 UTC by Report Bug Copy Issue Report Specify string format arguments as logging function parameters Loading history...
94			'triggers is disabled, skipping validation.' % (trigger_type_ref))
95			return None
96
97			cleaned = util_schema.validate(instance=parameters, schema=parameters_schema,
98			cls=util_schema.CustomValidator, use_default=True,
99			allow_default_none=True)
100
101			# Additional validation for CronTimer trigger
102			# TODO: If we need to add more checks like this we should consider abstracting this out.
103			if trigger_type_ref == CRON_TIMER_TRIGGER_REF:
104			# Validate that the user provided parameters are valid. This is required since JSON schema
105			# allows arbitrary strings, but not any arbitrary string is a valid CronTrigger argument
106			# Note: Constructor throws ValueError on invalid parameters
107			try:
108			CronTrigger(**parameters)
109			except:
110			msg = ('Cron trigger parameters do not match expected format.')
111			LOG.exception(msg)
112			raise TriggerParametersValidationException(msg)
113
114			return cleaned
115
116
117			def validate_trigger_payload(trigger_type_ref, payload):
118			"""
119			This function validates trigger payload parameters for system and user-defined triggers.
120
121			:param trigger_type_ref: Reference of a trigger type.
122			:type trigger_type_ref: ``str``
123
124			:param payload: Trigger payload.
125			:type payload: ``dict``
126
127			:return: Cleaned payload on success, None if validation is not performed.
128			"""
129			if not trigger_type_ref:
130			return None
131
132			is_system_trigger = trigger_type_ref in SYSTEM_TRIGGER_TYPES
133			if is_system_trigger:
134			# System trigger
135			payload_schema = SYSTEM_TRIGGER_TYPES[trigger_type_ref]['payload_schema']
136			else:
137			trigger_type_db = triggers.get_trigger_type_db(trigger_type_ref)
138			if not trigger_type_db:
139			# Trigger doesn't exist in the database
140			return None
141
142			payload_schema = getattr(trigger_type_db, 'payload_schema', {})
143			if not payload_schema:
144			# Payload schema not defined for the this trigger
145			return None
146
147			# We only validate non-system triggers if config option is set (enabled)
148			if not is_system_trigger and not cfg.CONF.system.validate_trigger_payload:
149			LOG.debug('Got non-system trigger "%s", but trigger payload validation for non-system'
			0 ignored issues – show Coding Style Best Practice introduced 2016-12-27 15:05 UTC by Report Bug Copy Issue Report Specify string format arguments as logging function parameters Loading history...
150			'triggers is disabled, skipping validation.' % (trigger_type_ref))
151			return None
152
153			cleaned = util_schema.validate(instance=payload, schema=payload_schema,
154			cls=util_schema.CustomValidator, use_default=True,
155			allow_default_none=True)
156
157			return cleaned
158

StackStorm / st2

Pull Request — master (#3804)

validate_trigger_parameters() C

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like