Passed
Pull Request — master (#3804)
by Lakshmi
05:58
created

validate_trigger_parameters()   C

Complexity

Conditions 9

Size

Total Lines 55

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   

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

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 by
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" % str(e)
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 by
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