/
action.py
127 lines (104 loc) · 5.15 KB
/
action.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
# 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 st2common.exceptions.apivalidation import ValueValidationException
from st2common.exceptions.db import StackStormDBObjectNotFoundError
from st2common import log as logging
from st2common.util.action_db import get_runnertype_by_name
from st2common.util import schema as util_schema
from st2common.content.utils import get_packs_base_paths
from st2common.content.utils import check_pack_content_directory_exists
from st2common.models.system.common import ResourceReference
from six.moves import range
__all__ = [
'validate_action',
'get_runner_model'
]
LOG = logging.getLogger(__name__)
def validate_action(action_api, runner_type_db=None):
"""
:param runner_type_db: RunnerTypeDB object belonging to this action. If not provided, it's
retrieved from the database.
:type runner_type_db: :class:`RunnerTypeDB`
"""
if not runner_type_db:
runner_db = get_runner_model(action_api)
else:
runner_db = runner_type_db
# Check if pack is valid.
if not _is_valid_pack(action_api.pack):
packs_base_paths = get_packs_base_paths()
packs_base_paths = ','.join(packs_base_paths)
msg = ('Content pack "%s" is not found or doesn\'t contain actions directory. '
'Searched in: %s' %
(action_api.pack, packs_base_paths))
raise ValueValidationException(msg)
# Check if parameters defined are valid.
action_ref = ResourceReference.to_string_reference(pack=action_api.pack, name=action_api.name)
_validate_parameters(action_ref, action_api.parameters, runner_db.runner_parameters)
def get_runner_model(action_api):
runner_db = None
# Check if runner exists.
try:
runner_db = get_runnertype_by_name(action_api.runner_type)
except StackStormDBObjectNotFoundError:
msg = 'RunnerType %s is not found.' % action_api.runner_type
raise ValueValidationException(msg)
return runner_db
def _is_valid_pack(pack):
return check_pack_content_directory_exists(pack=pack, content_type='actions')
def _validate_parameters(action_ref, action_params=None, runner_params=None):
position_params = {}
for action_param, action_param_meta in six.iteritems(action_params):
# Check if overridden runner parameters are permitted.
if action_param in runner_params:
for action_param_attr, value in six.iteritems(action_param_meta):
util_schema.validate_runner_parameter_attribute_override(
action_ref, action_param, action_param_attr,
value, runner_params[action_param].get(action_param_attr))
if 'position' in action_param_meta:
pos = action_param_meta['position']
param = position_params.get(pos, None)
if param:
msg = ('Parameters %s and %s have same position %d.' % (action_param, param, pos) +
' Position values have to be unique.')
raise ValueValidationException(msg)
else:
position_params[pos] = action_param
if 'immutable' in action_param_meta:
if action_param in runner_params:
runner_param_meta = runner_params[action_param]
if 'immutable' in runner_param_meta:
msg = 'Param %s is declared immutable in runner. ' % action_param + \
'Cannot override in action.'
raise ValueValidationException(msg)
if 'default' not in action_param_meta and 'default' not in runner_param_meta:
msg = 'Immutable param %s requires a default value.' % action_param
raise ValueValidationException(msg)
else:
if 'default' not in action_param_meta:
msg = 'Immutable param %s requires a default value.' % action_param
raise ValueValidationException(msg)
return _validate_position_values_contiguous(position_params)
def _validate_position_values_contiguous(position_params):
if not position_params:
return True
positions = sorted(position_params.keys())
contiguous = (positions == list(range(min(positions), max(positions) + 1)))
if not contiguous:
msg = 'Positions supplied %s for parameters are not contiguous.' % positions
raise ValueValidationException(msg)
return True