main() - Code Metrics - Inspection of "Add pytests to circle.yml" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#2598)

by Manas

created 2016-04-11 19:43 UTC

main() D

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	9
dl	0
loc	73
rs	4.1001

How to fix Long Method

#!/usr/bin/env python
# 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.

"""
Script which creates graphviz visualization of an action-chain workflow.
"""

import os
import argparse
import sets

try:
    from graphviz import Digraph
except ImportError:
    msg = ('Missing "graphviz" dependency. You can install it using pip: \n'
           'pip install graphviz')
    raise ImportError(msg)

from st2common.content.loader import MetaLoader
from st2actions.runners.actionchainrunner import ChainHolder


def main(metadata_path, output_path, print_source=False):
    metadata_path = os.path.abspath(metadata_path)
    metadata_dir = os.path.dirname(metadata_path)

    meta_loader = MetaLoader()
    data = meta_loader.load(metadata_path)

    action_name = data['name']
    entry_point = data['entry_point']

    workflow_metadata_path = os.path.join(metadata_dir, entry_point)
    chainspec = meta_loader.load(workflow_metadata_path)

    chain_holder = ChainHolder(chainspec, 'workflow')

    graph_label = '%s action-chain workflow visualization' % (action_name)

    graph_attr = {
        'rankdir': 'TD',
        'labelloc': 't',
        'fontsize': '15',
        'label': graph_label
    }
    node_attr = {}
    dot = Digraph(comment='Action chain work-flow visualization',
                  node_attr=node_attr, graph_attr=graph_attr, format='png')
    #  dot.body.extend(['rankdir=TD', 'size="10,5"'])

    # Add all nodes
    node = chain_holder.get_next_node()
    while node:
        dot.node(node.name, node.name)
        node = chain_holder.get_next_node(curr_node_name=node.name)

    # Add connections
    node = chain_holder.get_next_node()
    processed_nodes = sets.Set([node.name])
    nodes = [node]
    while nodes:
        previous_node = nodes.pop()
        success_node = chain_holder.get_next_node(curr_node_name=previous_node.name,
                                                  condition='on-success')
        failure_node = chain_holder.get_next_node(curr_node_name=previous_node.name,
                                                  condition='on-failure')

        # Add success node (if any)
        if success_node:
            dot.edge(previous_node.name, success_node.name, constraint='true',
                     color='green', label='on success')
            if success_node.name not in processed_nodes:
                nodes.append(success_node)
                processed_nodes.add(success_node.name)

        # Add failure node (if any)
        if failure_node:
            dot.edge(previous_node.name, failure_node.name, constraint='true',
                     color='red', label='on failure')
            if failure_node.name not in processed_nodes:
                nodes.append(failure_node)
                processed_nodes.add(failure_node.name)

    if print_source:
        print(dot.source)

    if output_path:
        output_path = os.path.join(output_path, action_name)
    else:
        output_path = output_path or os.path.join(os.getcwd(), action_name)

    dot.format = 'png'
    dot.render(output_path)

    print('Graph saved at %s' % (output_path + '.png'))

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Action chain visualization')
    parser.add_argument('--metadata-path', action='store', required=True,
                        help='Path to the workflow action metadata file')
    parser.add_argument('--output-path', action='store', required=False,
                        help='Output directory for the generated image')
    parser.add_argument('--print-source', action='store_true', default=False,
                        help='Print graphviz source code to the stdout')
    args = parser.parse_args()

    main(metadata_path=args.metadata_path, output_path=args.output_path,
         print_source=args.print_source)


1			#!/usr/bin/env python
2			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
3			# contributor license agreements. See the NOTICE file distributed with
4			# this work for additional information regarding copyright ownership.
5			# The ASF licenses this file to You under the Apache License, Version 2.0
6			# (the "License"); you may not use this file except in compliance with
7			# the License. You may obtain a copy of the License at
8			#
9			# http://www.apache.org/licenses/LICENSE-2.0
10			#
11			# Unless required by applicable law or agreed to in writing, software
12			# distributed under the License is distributed on an "AS IS" BASIS,
13			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14			# See the License for the specific language governing permissions and
15			# limitations under the License.
16
17			"""
18			Script which creates graphviz visualization of an action-chain workflow.
19			"""
20
21			import os
22			import argparse
23			import sets
24
25			try:
26			from graphviz import Digraph
27			except ImportError:
28			msg = ('Missing "graphviz" dependency. You can install it using pip: \n'
29			'pip install graphviz')
30			raise ImportError(msg)
31
32			from st2common.content.loader import MetaLoader
33			from st2actions.runners.actionchainrunner import ChainHolder
34
35
36			def main(metadata_path, output_path, print_source=False):
37			metadata_path = os.path.abspath(metadata_path)
38			metadata_dir = os.path.dirname(metadata_path)
39
40			meta_loader = MetaLoader()
41			data = meta_loader.load(metadata_path)
42
43			action_name = data['name']
44			entry_point = data['entry_point']
45
46			workflow_metadata_path = os.path.join(metadata_dir, entry_point)
47			chainspec = meta_loader.load(workflow_metadata_path)
48
49			chain_holder = ChainHolder(chainspec, 'workflow')
50
51			graph_label = '%s action-chain workflow visualization' % (action_name)
52
53			graph_attr = {
54			'rankdir': 'TD',
55			'labelloc': 't',
56			'fontsize': '15',
57			'label': graph_label
58			}
59			node_attr = {}
60			dot = Digraph(comment='Action chain work-flow visualization',
61			node_attr=node_attr, graph_attr=graph_attr, format='png')
62			# dot.body.extend(['rankdir=TD', 'size="10,5"'])
63
64			# Add all nodes
65			node = chain_holder.get_next_node()
66			while node:
67			dot.node(node.name, node.name)
68			node = chain_holder.get_next_node(curr_node_name=node.name)
69
70			# Add connections
71			node = chain_holder.get_next_node()
72			processed_nodes = sets.Set([node.name])
73			nodes = [node]
74			while nodes:
75			previous_node = nodes.pop()
76			success_node = chain_holder.get_next_node(curr_node_name=previous_node.name,
77			condition='on-success')
78			failure_node = chain_holder.get_next_node(curr_node_name=previous_node.name,
79			condition='on-failure')
80
81			# Add success node (if any)
82			if success_node:
83			dot.edge(previous_node.name, success_node.name, constraint='true',
84			color='green', label='on success')
85			if success_node.name not in processed_nodes:
86			nodes.append(success_node)
87			processed_nodes.add(success_node.name)
88
89			# Add failure node (if any)
90			if failure_node:
91			dot.edge(previous_node.name, failure_node.name, constraint='true',
92			color='red', label='on failure')
93			if failure_node.name not in processed_nodes:
94			nodes.append(failure_node)
95			processed_nodes.add(failure_node.name)
96
97			if print_source:
98			print(dot.source)
99
100			if output_path:
101			output_path = os.path.join(output_path, action_name)
102			else:
103			output_path = output_path or os.path.join(os.getcwd(), action_name)
104
105			dot.format = 'png'
106			dot.render(output_path)
107
108			print('Graph saved at %s' % (output_path + '.png'))
109
110			if __name__ == '__main__':
111			parser = argparse.ArgumentParser(description='Action chain visualization')
112			parser.add_argument('--metadata-path', action='store', required=True,
113			help='Path to the workflow action metadata file')
114			parser.add_argument('--output-path', action='store', required=False,
115			help='Output directory for the generated image')
116			parser.add_argument('--print-source', action='store_true', default=False,
117			help='Print graphviz source code to the stdout')
118			args = parser.parse_args()
119
120			main(metadata_path=args.metadata_path, output_path=args.output_path,
121			print_source=args.print_source)
122

StackStorm / st2

Pull Request — master (#2598)

main() D

Complexity

Size

Duplication

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like