ContentPackLoader._get_aliases() - Code Metrics - Inspection of "Fix wrong information for /api/v1/apikeys/ doc" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#4194)

by Arma

created 2018-06-24 10:28 UTC

ContentPackLoader._get_aliases() A

↳ Parent: ContentPackLoader

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# 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 os
import re

from yaml.parser import ParserError
import six

from st2common import log as logging
from st2common.constants.meta import ALLOWED_EXTS
from st2common.constants.meta import PARSER_FUNCS
from st2common.constants.pack import MANIFEST_FILE_NAME
from st2common.constants.runners import MANIFEST_FILE_NAME as RUNNER_MANIFEST_FILE_NAME
from st2common.constants.runners import RUNNER_NAME_WHITELIST

__all__ = [
    'RunnersLoader',
    'ContentPackLoader',
    'MetaLoader'
]

LOG = logging.getLogger(__name__)


class RunnersLoader(object):
    """Class for loading runners from directories on disk.
    """

    def get_runners(self, base_dirs):
        """Retrieve a list of runners in the provided directories.

        :return: Dictionary where the key is runner name and the value is full path to the runner
                 directory.
        :rtype: ``dict``
        """
        assert isinstance(base_dirs, list)

        result = {}
        for base_dir in base_dirs:
            if not os.path.isdir(base_dir):
                raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))

            runners_in_dir = self._get_runners_from_dir(base_dir=base_dir)
            result.update(runners_in_dir)

        return result

    def _get_runners_from_dir(self, base_dir):
        result = {}
        for runner_name in os.listdir(base_dir):
            runner_dir = os.path.join(base_dir, runner_name)
            runner_manifest_file = os.path.join(runner_dir, RUNNER_MANIFEST_FILE_NAME)

            if os.path.isdir(runner_dir) and os.path.isfile(runner_manifest_file):
                result[runner_name] = runner_dir

        return result


class ContentPackLoader(object):
    """
    Class for loading pack and pack content information from directories on disk.
    """

    # TODO: Rename "get_content" methods since they don't actually return
    # content - they just return a path

    ALLOWED_CONTENT_TYPES = [
        'triggers',
        'sensors',
        'actions',
        'rules',
        'aliases',
        'policies'
    ]

    def get_packs(self, base_dirs):
        """
        Retrieve a list of packs in the provided directories.

        :return: Dictionary where the key is pack name and the value is full path to the pack
                 directory.
        :rtype: ``dict``
        """
        assert isinstance(base_dirs, list)

        result = {}
        for base_dir in base_dirs:
            if not os.path.isdir(base_dir):
                raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))

            packs_in_dir = self._get_packs_from_dir(base_dir=base_dir)
            result.update(packs_in_dir)

        return result

    def get_content(self, base_dirs, content_type):
        """
        Retrieve content from the provided directories.

        Provided directories are searched from left to right. If a pack with the same name exists
        in multiple directories, first pack which is found wins.

        :param base_dirs: Directories to look into.
        :type base_dirs: ``list``

        :param content_type: Content type to look for (sensors, actions, rules).
        :type content_type: ``str``

        :rtype: ``dict``
        """
        assert isinstance(base_dirs, list)

        if content_type not in self.ALLOWED_CONTENT_TYPES:
            raise ValueError('Unsupported content_type: %s' % (content_type))

        content = {}
        pack_to_dir_map = {}
        for base_dir in base_dirs:
            if not os.path.isdir(base_dir):
                raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))

            dir_content = self._get_content_from_dir(base_dir=base_dir, content_type=content_type)

            # Check for duplicate packs
            for pack_name, pack_content in six.iteritems(dir_content):
                if pack_name in content:
                    pack_dir = pack_to_dir_map[pack_name]
                    LOG.warning('Pack "%s" already found in "%s", ignoring content from "%s"' %

                                (pack_name, pack_dir, base_dir))
                else:
                    content[pack_name] = pack_content
                    pack_to_dir_map[pack_name] = base_dir

        return content

    def get_content_from_pack(self, pack_dir, content_type):
        """
        Retrieve content from the provided pack directory.

        :param pack_dir: Path to the pack directory.
        :type pack_dir: ``str``

        :param content_type: Content type to look for (sensors, actions, rules).
        :type content_type: ``str``

        :rtype: ``str``
        """
        if content_type not in self.ALLOWED_CONTENT_TYPES:
            raise ValueError('Unsupported content_type: %s' % (content_type))

        if not os.path.isdir(pack_dir):
            raise ValueError('Directory "%s" doesn\'t exist' % (pack_dir))

        content = self._get_content_from_pack_dir(pack_dir=pack_dir,
                                                  content_type=content_type)
        return content

    def _get_packs_from_dir(self, base_dir):
        result = {}
        for pack_name in os.listdir(base_dir):
            pack_dir = os.path.join(base_dir, pack_name)
            pack_manifest_file = os.path.join(pack_dir, MANIFEST_FILE_NAME)

            if os.path.isdir(pack_dir) and os.path.isfile(pack_manifest_file):
                result[pack_name] = pack_dir

        return result

    def _get_content_from_dir(self, base_dir, content_type):
        content = {}
        for pack in os.listdir(base_dir):
            # TODO: Use function from util which escapes the name
            pack_dir = os.path.join(base_dir, pack)

            # Ignore missing or non directories
            try:
                pack_content = self._get_content_from_pack_dir(pack_dir=pack_dir,
                                                               content_type=content_type)
            except ValueError:
                continue
            else:
                content[pack] = pack_content

        return content

    def _get_content_from_pack_dir(self, pack_dir, content_type):
        content_types = dict(
            triggers=self._get_triggers,
            sensors=self._get_sensors,
            actions=self._get_actions,
            rules=self._get_rules,
            aliases=self._get_aliases,
            policies=self._get_policies
        )

        get_func = content_types.get(content_type)

        if get_func is None:
            raise ValueError('Invalid content_type: %s' % (content_type))

        if not os.path.isdir(pack_dir):
            raise ValueError('Directory "%s" doesn\'t exist' % (pack_dir))

        pack_content = get_func(pack_dir=pack_dir)
        return pack_content

    def _get_triggers(self, pack_dir):
        return self._get_folder(pack_dir=pack_dir, content_type='triggers')

    def _get_sensors(self, pack_dir):
        return self._get_folder(pack_dir=pack_dir, content_type='sensors')

    def _get_actions(self, pack_dir):
        return self._get_folder(pack_dir=pack_dir, content_type='actions')

    def _get_rules(self, pack_dir):
        return self._get_folder(pack_dir=pack_dir, content_type='rules')

    def _get_aliases(self, pack_dir):
        return self._get_folder(pack_dir=pack_dir, content_type='aliases')

    def _get_policies(self, pack_dir):
        return self._get_folder(pack_dir=pack_dir, content_type='policies')

    def _get_folder(self, pack_dir, content_type):
        path = os.path.join(pack_dir, content_type)
        if not os.path.isdir(path):
            return None
        return path

    def _get_runners_from_dir(self, base_dir):
        result = {}
        for runner_name in os.listdir(base_dir):
            if not re.match(RUNNER_NAME_WHITELIST, runner_name):
                raise ValueError('Invalid runner name "%s"' % (runner_name))

            runner_dir = os.path.join(base_dir, runner_name)
            runner_manifest_file = os.path.join(runner_dir, RUNNER_MANIFEST_FILE_NAME)

            if os.path.isdir(runner_dir) and os.path.isfile(runner_manifest_file):
                LOG.debug("Loading runner manifest for: %s" % (runner_name))

                result[runner_name] = runner_dir
            else:
                LOG.debug("Could not load manifest for runner: %s" % (runner_name))


        return result

    def get_runners(self, base_dirs):
        """Retrieve a list of runners in the provided directories.

        :return: Dictionary where the key is runner name and the value is full path to the runner
                 directory.
        :rtype: ``dict``
        """
        assert isinstance(base_dirs, list)

        result = {}
        for base_dir in base_dirs:
            if not os.path.isdir(base_dir):
                raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))

            runners_in_dir = self._get_runners_from_dir(base_dir=base_dir)
            result.update(runners_in_dir)

        return result


class MetaLoader(object):
    """
    Class for loading and parsing pack and resource metadata files.
    """

    def load(self, file_path, expected_type=None):
        """
        Loads content from file_path if file_path's extension
        is one of allowed ones (See ALLOWED_EXTS).

        Throws UnsupportedMetaException on disallowed filetypes.
        Throws ValueError on malformed meta.

        :param file_path: Absolute path to the file to load content from.
        :type file_path: ``str``

        :param expected_type: Expected type for the loaded and parsed content (optional).
        :type expected_type: ``object``

        :rtype: ``dict``
        """
        file_name, file_ext = os.path.splitext(file_path)

        if file_ext not in ALLOWED_EXTS:
            raise Exception('Unsupported meta type %s, file %s. Allowed: %s' %
                            (file_ext, file_path, ALLOWED_EXTS))

        result = self._load(PARSER_FUNCS[file_ext], file_path)

        if expected_type and not isinstance(result, expected_type):
            actual_type = type(result).__name__
            error = 'Expected "%s", got "%s"' % (expected_type.__name__, actual_type)
            raise ValueError(error)

        return result

    def _load(self, parser_func, file_path):
        with open(file_path, 'r') as fd:
            try:
                return parser_func(fd)
            except ValueError:
                LOG.exception('Failed loading content from %s.', file_path)
                raise
            except ParserError:
                LOG.exception('Failed loading content from %s.', file_path)
                raise


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 os
18			import re
19
20			from yaml.parser import ParserError
21			import six
22
23			from st2common import log as logging
24			from st2common.constants.meta import ALLOWED_EXTS
25			from st2common.constants.meta import PARSER_FUNCS
26			from st2common.constants.pack import MANIFEST_FILE_NAME
27			from st2common.constants.runners import MANIFEST_FILE_NAME as RUNNER_MANIFEST_FILE_NAME
28			from st2common.constants.runners import RUNNER_NAME_WHITELIST
29
30			__all__ = [
31			'RunnersLoader',
32			'ContentPackLoader',
33			'MetaLoader'
34			]
35
36			LOG = logging.getLogger(__name__)
37
38
39			class RunnersLoader(object):
40			"""Class for loading runners from directories on disk.
41			"""
42
43			def get_runners(self, base_dirs):
44			"""Retrieve a list of runners in the provided directories.
45
46			:return: Dictionary where the key is runner name and the value is full path to the runner
47			directory.
48			:rtype: ``dict``
49			"""
50			assert isinstance(base_dirs, list)
51
52			result = {}
53			for base_dir in base_dirs:
54			if not os.path.isdir(base_dir):
55			raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))
56
57			runners_in_dir = self._get_runners_from_dir(base_dir=base_dir)
58			result.update(runners_in_dir)
59
60			return result
61
62			def _get_runners_from_dir(self, base_dir):
63			result = {}
64			for runner_name in os.listdir(base_dir):
65			runner_dir = os.path.join(base_dir, runner_name)
66			runner_manifest_file = os.path.join(runner_dir, RUNNER_MANIFEST_FILE_NAME)
67
68			if os.path.isdir(runner_dir) and os.path.isfile(runner_manifest_file):
69			result[runner_name] = runner_dir
70
71			return result
72
73
74			class ContentPackLoader(object):
75			"""
76			Class for loading pack and pack content information from directories on disk.
77			"""
78
79			# TODO: Rename "get_content" methods since they don't actually return
80			# content - they just return a path
81
82			ALLOWED_CONTENT_TYPES = [
83			'triggers',
84			'sensors',
85			'actions',
86			'rules',
87			'aliases',
88			'policies'
89			]
90
91			def get_packs(self, base_dirs):
92			"""
93			Retrieve a list of packs in the provided directories.
94
95			:return: Dictionary where the key is pack name and the value is full path to the pack
96			directory.
97			:rtype: ``dict``
98			"""
99			assert isinstance(base_dirs, list)
100
101			result = {}
102			for base_dir in base_dirs:
103			if not os.path.isdir(base_dir):
104			raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))
105
106			packs_in_dir = self._get_packs_from_dir(base_dir=base_dir)
107			result.update(packs_in_dir)
108
109			return result
110
111			def get_content(self, base_dirs, content_type):
112			"""
113			Retrieve content from the provided directories.
114
115			Provided directories are searched from left to right. If a pack with the same name exists
116			in multiple directories, first pack which is found wins.
117
118			:param base_dirs: Directories to look into.
119			:type base_dirs: ``list``
120
121			:param content_type: Content type to look for (sensors, actions, rules).
122			:type content_type: ``str``
123
124			:rtype: ``dict``
125			"""
126			assert isinstance(base_dirs, list)
127
128			if content_type not in self.ALLOWED_CONTENT_TYPES:
129			raise ValueError('Unsupported content_type: %s' % (content_type))
130
131			content = {}
132			pack_to_dir_map = {}
133			for base_dir in base_dirs:
134			if not os.path.isdir(base_dir):
135			raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))
136
137			dir_content = self._get_content_from_dir(base_dir=base_dir, content_type=content_type)
138
139			# Check for duplicate packs
140			for pack_name, pack_content in six.iteritems(dir_content):
141			if pack_name in content:
142			pack_dir = pack_to_dir_map[pack_name]
143			LOG.warning('Pack "%s" already found in "%s", ignoring content from "%s"' %
			0 ignored issues – show Coding Style Best Practice introduced 2015-11-20 23:04 UTC by Report Bug Copy Issue Report Specify string format arguments as logging function parameters Loading history...
144			(pack_name, pack_dir, base_dir))
145			else:
146			content[pack_name] = pack_content
147			pack_to_dir_map[pack_name] = base_dir
148
149			return content
150
151			def get_content_from_pack(self, pack_dir, content_type):
152			"""
153			Retrieve content from the provided pack directory.
154
155			:param pack_dir: Path to the pack directory.
156			:type pack_dir: ``str``
157
158			:param content_type: Content type to look for (sensors, actions, rules).
159			:type content_type: ``str``
160
161			:rtype: ``str``
162			"""
163			if content_type not in self.ALLOWED_CONTENT_TYPES:
164			raise ValueError('Unsupported content_type: %s' % (content_type))
165
166			if not os.path.isdir(pack_dir):
167			raise ValueError('Directory "%s" doesn\'t exist' % (pack_dir))
168
169			content = self._get_content_from_pack_dir(pack_dir=pack_dir,
170			content_type=content_type)
171			return content
172
173			def _get_packs_from_dir(self, base_dir):
174			result = {}
175			for pack_name in os.listdir(base_dir):
176			pack_dir = os.path.join(base_dir, pack_name)
177			pack_manifest_file = os.path.join(pack_dir, MANIFEST_FILE_NAME)
178
179			if os.path.isdir(pack_dir) and os.path.isfile(pack_manifest_file):
180			result[pack_name] = pack_dir
181
182			return result
183
184			def _get_content_from_dir(self, base_dir, content_type):
185			content = {}
186			for pack in os.listdir(base_dir):
187			# TODO: Use function from util which escapes the name
188			pack_dir = os.path.join(base_dir, pack)
189
190			# Ignore missing or non directories
191			try:
192			pack_content = self._get_content_from_pack_dir(pack_dir=pack_dir,
193			content_type=content_type)
194			except ValueError:
195			continue
196			else:
197			content[pack] = pack_content
198
199			return content
200
201			def _get_content_from_pack_dir(self, pack_dir, content_type):
202			content_types = dict(
203			triggers=self._get_triggers,
204			sensors=self._get_sensors,
205			actions=self._get_actions,
206			rules=self._get_rules,
207			aliases=self._get_aliases,
208			policies=self._get_policies
209			)
210
211			get_func = content_types.get(content_type)
212
213			if get_func is None:
214			raise ValueError('Invalid content_type: %s' % (content_type))
215
216			if not os.path.isdir(pack_dir):
217			raise ValueError('Directory "%s" doesn\'t exist' % (pack_dir))
218
219			pack_content = get_func(pack_dir=pack_dir)
220			return pack_content
221
222			def _get_triggers(self, pack_dir):
223			return self._get_folder(pack_dir=pack_dir, content_type='triggers')
224
225			def _get_sensors(self, pack_dir):
226			return self._get_folder(pack_dir=pack_dir, content_type='sensors')
227
228			def _get_actions(self, pack_dir):
229			return self._get_folder(pack_dir=pack_dir, content_type='actions')
230
231			def _get_rules(self, pack_dir):
232			return self._get_folder(pack_dir=pack_dir, content_type='rules')
233
234			def _get_aliases(self, pack_dir):
235			return self._get_folder(pack_dir=pack_dir, content_type='aliases')
236
237			def _get_policies(self, pack_dir):
238			return self._get_folder(pack_dir=pack_dir, content_type='policies')
239
240			def _get_folder(self, pack_dir, content_type):
241			path = os.path.join(pack_dir, content_type)
242			if not os.path.isdir(path):
243			return None
244			return path
245
246			def _get_runners_from_dir(self, base_dir):
247			result = {}
248			for runner_name in os.listdir(base_dir):
249			if not re.match(RUNNER_NAME_WHITELIST, runner_name):
250			raise ValueError('Invalid runner name "%s"' % (runner_name))
251
252			runner_dir = os.path.join(base_dir, runner_name)
253			runner_manifest_file = os.path.join(runner_dir, RUNNER_MANIFEST_FILE_NAME)
254
255			if os.path.isdir(runner_dir) and os.path.isfile(runner_manifest_file):
256			LOG.debug("Loading runner manifest for: %s" % (runner_name))
			0 ignored issues – show Coding Style Best Practice introduced 2017-04-18 08:10 UTC by Report Bug Copy Issue Report Specify string format arguments as logging function parameters Loading history...
257			result[runner_name] = runner_dir
258			else:
259			LOG.debug("Could not load manifest for runner: %s" % (runner_name))
			0 ignored issues – show Coding Style Best Practice introduced 2016-09-28 00:35 UTC by Report Bug Copy Issue Report Specify string format arguments as logging function parameters Loading history...
260
261			return result
262
263			def get_runners(self, base_dirs):
264			"""Retrieve a list of runners in the provided directories.
265
266			:return: Dictionary where the key is runner name and the value is full path to the runner
267			directory.
268			:rtype: ``dict``
269			"""
270			assert isinstance(base_dirs, list)
271
272			result = {}
273			for base_dir in base_dirs:
274			if not os.path.isdir(base_dir):
275			raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))
276
277			runners_in_dir = self._get_runners_from_dir(base_dir=base_dir)
278			result.update(runners_in_dir)
279
280			return result
281
282
283			class MetaLoader(object):
284			"""
285			Class for loading and parsing pack and resource metadata files.
286			"""
287
288			def load(self, file_path, expected_type=None):
289			"""
290			Loads content from file_path if file_path's extension
291			is one of allowed ones (See ALLOWED_EXTS).
292
293			Throws UnsupportedMetaException on disallowed filetypes.
294			Throws ValueError on malformed meta.
295
296			:param file_path: Absolute path to the file to load content from.
297			:type file_path: ``str``
298
299			:param expected_type: Expected type for the loaded and parsed content (optional).
300			:type expected_type: ``object``
301
302			:rtype: ``dict``
303			"""
304			file_name, file_ext = os.path.splitext(file_path)
305
306			if file_ext not in ALLOWED_EXTS:
307			raise Exception('Unsupported meta type %s, file %s. Allowed: %s' %
308			(file_ext, file_path, ALLOWED_EXTS))
309
310			result = self._load(PARSER_FUNCS[file_ext], file_path)
311
312			if expected_type and not isinstance(result, expected_type):
313			actual_type = type(result).__name__
314			error = 'Expected "%s", got "%s"' % (expected_type.__name__, actual_type)
315			raise ValueError(error)
316
317			return result
318
319			def _load(self, parser_func, file_path):
320			with open(file_path, 'r') as fd:
321			try:
322			return parser_func(fd)
323			except ValueError:
324			LOG.exception('Failed loading content from %s.', file_path)
325			raise
326			except ParserError:
327			LOG.exception('Failed loading content from %s.', file_path)
328			raise
329

StackStorm / st2

Pull Request — master (#4194)

ContentPackLoader._get_aliases() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like