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 2018-03-01 21:04 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	55
rs	5.4159

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.

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

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