request_values() - Code Metrics - Inspection of "Brocade VDX" - StackStorm/st2contrib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#543)

unknown

created 2016-08-24 17:48 UTC

request_values() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	7
c	1
b	0
f	0
dl	0
loc	32
rs	5.5

"""Test generator for VDX pack
"""
import os
import sys
import inspect
import imp
import xml.etree.ElementTree as ET
import yaml
from jinja2 import Template
ROOT_DIR = './pack/actions/'
ST2_REPO_PATH = "/tmp/st2/"
IGNORE_PARAMS = ['username', 'password', 'ip']
TYPICAL_VALUES = {
    'ip_addr': '10.10.0.1/24',
    'inter_type': 'tengigabitethernet',
    'rbridge_id': '224',
    'inter': '10/0/1',
    'vlan_id': '44',
    'vrf': 'test',
    'remote_as': '18003',
    'delete': False,
    'get': False
}


def get_yaml_files():
    """Parse directory and get all YAML files
    """
    yaml_files = []
    for subdir, _, action_files in os.walk(ROOT_DIR):
        for action_file in action_files:
            if 'yaml' in action_file:
                yaml_files.append("%s%s" % (subdir, action_file))

    return yaml_files


def parse_yaml_files(yaml_files):
    """Parse YAML files and add return dictionary of all parsed YAML files
    """
    result = {}
    for yaml_file in yaml_files:
        with open(yaml_file, 'r') as yaml_fd:
            yaml_output = yaml.load(yaml_fd.read())
            result.update({yaml_output['name']: yaml_output})

    return result


def request_user_bool(prompt):
    """Request a yes or no from the user and return the result as a bool
    """
    user_input = raw_input(prompt)
    if user_input is not "y" and user_input is not "n":
        print 'Incorrect entry. Pleasae use "y" or "n".'
        return request_user_bool(prompt)
    else:
        return bool(user_input == "y")


def request_values(actions):
    """Given a dictionary of actions look at its parameters and request
    values for parameters without a default value. This is returned as a
    dictionary adding the `value` filed to the action's dictionary."""
    for action_name, action_metadata in actions.iteritems():
        print "New Action: %s" % action_name
        for parameter, parameter_metadata in\
                action_metadata['parameters'].iteritems():
            if parameter in IGNORE_PARAMS:
                continue

            typical_value = TYPICAL_VALUES.get(parameter, None)
            if typical_value is not None:
                if request_user_bool("  Use typical value (%s) for %s? (y/n):"
                                     % (typical_value, parameter)):
                    actions[action_name]['parameters'][parameter].update(
                        value=typical_value
                    )
                    continue

            default = parameter_metadata.pop('default', None)
            if default is None:
                user_value = raw_input('  Enter value for %s: ' % parameter)
                actions[action_name]['parameters'][parameter].update(
                    value=user_value
                )
            else:
                actions[action_name]['parameters'][parameter].update(
                    value=default
                )

    return actions


class CallbackClass(object):  # pylint:disable=too-few-public-methods
    """Callback class. Contains action_callback method to feed to tests and
    holds results for retriving later.
    """
    result = None

    def action_callback(self, xml, **kwargs):  # pylint:disable=unused-argument
        """Callback for action
        """
        self.result = ET.tostring(xml)


def validate_actions_result(actions, callback):
    """Run each action and confirm with user that the expected result was
    returned.
    """
    invalid_actions = []
    for action, action_metadata in actions.iteritems():
        action_module = imp.load_source(
            action,
            '%s%s' % (ROOT_DIR, action_metadata['entry_point'])
        )
        classes = inspect.getmembers(action_module, predicate=inspect.isclass)
        action_instance = classes[1][1]()
        kwargs = {}
        for parameter, parameter_metadata in\
                action_metadata['parameters'].iteritems():
            if parameter in IGNORE_PARAMS:
                kwargs[parameter] = ' '
            else:
                kwargs[parameter] = parameter_metadata['value']
        kwargs['test'] = True
        kwargs['callback'] = callback.action_callback
        action_instance.run(**kwargs)
        print callback.result
        action_metadata['expected_result'] = callback.result
        actions.update({action: action_metadata})
        if request_user_bool("Does the above output match what you expect?"
                             "(y/n): "):
            action_metadata['expected_result'] = callback.result
            actions.update({action: action_metadata})
        else:
            invalid_actions.append(action)

    return actions, invalid_actions


def generate_test_files(actions, invalid_actions):
    """Generate final tests and write them to disk.
    """
    with open('test_action_template.j2', 'r') as template_file:
        template = Template(template_file.read())
    for action, action_metadata in actions.iteritems():
        if action in invalid_actions:
            print "Skipping %s since its result was deemed invalid." % action
            continue

        expected_result = [
            action_metadata['expected_result'][i:i + 65]
            for i in range(0, len(action_metadata['expected_result']), 65)
        ]

        class_name = 'Test'

        for word in action.split('_'):
            class_name += word.capitalize()

        test_code = template.render(
            action_name=action,
            class_name=class_name,
            parameters=action_metadata['parameters'],
            expected_results=expected_result
        )
        with open("./pack/tests/test_action_%s" %
                  action_metadata['entry_point'], "w+") as test_fd:
            test_fd.write(test_code)


def print_invalid_actions(invalid_actions):
    """Print list of given invalid actions.
    """
    if len(invalid_actions) > 0:
        print "List of actions with incorrect result:"
        for action in invalid_actions:
            print action
    else:
        print "No invalid actions."


def setup():
    """Setup test build environment.
    """
    st2_artifacts = [
        'st2actions',
        'st2common'
    ]
    for artifact in st2_artifacts:
        sys.path.append(ST2_REPO_PATH + artifact)


def write_answer_file(answers):
    """Write answer file to disk.
    """
    with open("answerfile.yaml", "w+") as answer_file:
        answer_file.write(yaml.dump(answers))


def load_answer_file():
    """Write answer file to disk.
    """
    try:
        with open("answerfile.yaml", "r") as answer_file:
            return yaml.load(answer_file.read())
    except EnvironmentError:
        print "An error occoured trying to load the answer file. Continuing."
        return None


def main():
    """Main entry point
    """
    print "Setting up test build env."
    setup()
    print "Getting YAML files from disk."
    yaml_files = get_yaml_files()
    print "Parsing YAML files to get metadata."
    actions = parse_yaml_files(yaml_files)
    result = None
    if request_user_bool(
            "Would you like to attempt to load an answer file? (y/n): "
    ):
        result = load_answer_file()
    if result is None:
        print "Requesting values for actions without default value."
        actions = request_values(actions)
    else:
        actions = result
    if request_user_bool(
            "Would you like to write these answers to an answer"
            "file? (y/n): "
    ):
        write_answer_file(actions)
    print "Validating actions."
    callback = CallbackClass()
    actions, invalid_actions = validate_actions_result(actions, callback)
    print "Generating test files."
    generate_test_files(actions, invalid_actions)
    print_invalid_actions(invalid_actions)
    print "Test files have been generated."


if __name__ == "__main__":
    main()


1			"""Test generator for VDX pack
2			"""
3			import os
4			import sys
5			import inspect
6			import imp
7			import xml.etree.ElementTree as ET
8			import yaml
9			from jinja2 import Template
10			ROOT_DIR = './pack/actions/'
11			ST2_REPO_PATH = "/tmp/st2/"
12			IGNORE_PARAMS = ['username', 'password', 'ip']
13			TYPICAL_VALUES = {
14			'ip_addr': '10.10.0.1/24',
15			'inter_type': 'tengigabitethernet',
16			'rbridge_id': '224',
17			'inter': '10/0/1',
18			'vlan_id': '44',
19			'vrf': 'test',
20			'remote_as': '18003',
21			'delete': False,
22			'get': False
23			}
24
25
26			def get_yaml_files():
27			"""Parse directory and get all YAML files
28			"""
29			yaml_files = []
30			for subdir, _, action_files in os.walk(ROOT_DIR):
31			for action_file in action_files:
32			if 'yaml' in action_file:
33			yaml_files.append("%s%s" % (subdir, action_file))
34
35			return yaml_files
36
37
38			def parse_yaml_files(yaml_files):
39			"""Parse YAML files and add return dictionary of all parsed YAML files
40			"""
41			result = {}
42			for yaml_file in yaml_files:
43			with open(yaml_file, 'r') as yaml_fd:
44			yaml_output = yaml.load(yaml_fd.read())
45			result.update({yaml_output['name']: yaml_output})
46
47			return result
48
49
50			def request_user_bool(prompt):
51			"""Request a yes or no from the user and return the result as a bool
52			"""
53			user_input = raw_input(prompt)
54			if user_input is not "y" and user_input is not "n":
55			print 'Incorrect entry. Pleasae use "y" or "n".'
56			return request_user_bool(prompt)
57			else:
58			return bool(user_input == "y")
59
60
61			def request_values(actions):
62			"""Given a dictionary of actions look at its parameters and request
63			values for parameters without a default value. This is returned as a
64			dictionary adding the `value` filed to the action's dictionary."""
65			for action_name, action_metadata in actions.iteritems():
66			print "New Action: %s" % action_name
67			for parameter, parameter_metadata in\
68			action_metadata['parameters'].iteritems():
69			if parameter in IGNORE_PARAMS:
70			continue
71
72			typical_value = TYPICAL_VALUES.get(parameter, None)
73			if typical_value is not None:
74			if request_user_bool(" Use typical value (%s) for %s? (y/n):"
75			% (typical_value, parameter)):
76			actions[action_name]['parameters'][parameter].update(
77			value=typical_value
78			)
79			continue
80
81			default = parameter_metadata.pop('default', None)
82			if default is None:
83			user_value = raw_input(' Enter value for %s: ' % parameter)
84			actions[action_name]['parameters'][parameter].update(
85			value=user_value
86			)
87			else:
88			actions[action_name]['parameters'][parameter].update(
89			value=default
90			)
91
92			return actions
93
94
95			class CallbackClass(object): # pylint:disable=too-few-public-methods
96			"""Callback class. Contains action_callback method to feed to tests and
97			holds results for retriving later.
98			"""
99			result = None
100
101			def action_callback(self, xml, **kwargs): # pylint:disable=unused-argument
102			"""Callback for action
103			"""
104			self.result = ET.tostring(xml)
105
106
107			def validate_actions_result(actions, callback):
108			"""Run each action and confirm with user that the expected result was
109			returned.
110			"""
111			invalid_actions = []
112			for action, action_metadata in actions.iteritems():
113			action_module = imp.load_source(
114			action,
115			'%s%s' % (ROOT_DIR, action_metadata['entry_point'])
116			)
117			classes = inspect.getmembers(action_module, predicate=inspect.isclass)
118			action_instance = classes[1][1]()
119			kwargs = {}
120			for parameter, parameter_metadata in\
121			action_metadata['parameters'].iteritems():
122			if parameter in IGNORE_PARAMS:
123			kwargs[parameter] = ' '
124			else:
125			kwargs[parameter] = parameter_metadata['value']
126			kwargs['test'] = True
127			kwargs['callback'] = callback.action_callback
128			action_instance.run(**kwargs)
129			print callback.result
130			action_metadata['expected_result'] = callback.result
131			actions.update({action: action_metadata})
132			if request_user_bool("Does the above output match what you expect?"
133			"(y/n): "):
134			action_metadata['expected_result'] = callback.result
135			actions.update({action: action_metadata})
136			else:
137			invalid_actions.append(action)
138
139			return actions, invalid_actions
140
141
142			def generate_test_files(actions, invalid_actions):
143			"""Generate final tests and write them to disk.
144			"""
145			with open('test_action_template.j2', 'r') as template_file:
146			template = Template(template_file.read())
147			for action, action_metadata in actions.iteritems():
148			if action in invalid_actions:
149			print "Skipping %s since its result was deemed invalid." % action
150			continue
151
152			expected_result = [
153			action_metadata['expected_result'][i:i + 65]
154			for i in range(0, len(action_metadata['expected_result']), 65)
155			]
156
157			class_name = 'Test'
158
159			for word in action.split('_'):
160			class_name += word.capitalize()
161
162			test_code = template.render(
163			action_name=action,
164			class_name=class_name,
165			parameters=action_metadata['parameters'],
166			expected_results=expected_result
167			)
168			with open("./pack/tests/test_action_%s" %
169			action_metadata['entry_point'], "w+") as test_fd:
170			test_fd.write(test_code)
171
172
173			def print_invalid_actions(invalid_actions):
174			"""Print list of given invalid actions.
175			"""
176			if len(invalid_actions) > 0:
177			print "List of actions with incorrect result:"
178			for action in invalid_actions:
179			print action
180			else:
181			print "No invalid actions."
182
183
184			def setup():
185			"""Setup test build environment.
186			"""
187			st2_artifacts = [
188			'st2actions',
189			'st2common'
190			]
191			for artifact in st2_artifacts:
192			sys.path.append(ST2_REPO_PATH + artifact)
193
194
195			def write_answer_file(answers):
196			"""Write answer file to disk.
197			"""
198			with open("answerfile.yaml", "w+") as answer_file:
199			answer_file.write(yaml.dump(answers))
200
201
202			def load_answer_file():
203			"""Write answer file to disk.
204			"""
205			try:
206			with open("answerfile.yaml", "r") as answer_file:
207			return yaml.load(answer_file.read())
208			except EnvironmentError:
209			print "An error occoured trying to load the answer file. Continuing."
210			return None
211
212
213			def main():
214			"""Main entry point
215			"""
216			print "Setting up test build env."
217			setup()
218			print "Getting YAML files from disk."
219			yaml_files = get_yaml_files()
220			print "Parsing YAML files to get metadata."
221			actions = parse_yaml_files(yaml_files)
222			result = None
223			if request_user_bool(
224			"Would you like to attempt to load an answer file? (y/n): "
225			):
226			result = load_answer_file()
227			if result is None:
228			print "Requesting values for actions without default value."
229			actions = request_values(actions)
230			else:
231			actions = result
232			if request_user_bool(
233			"Would you like to write these answers to an answer"
234			"file? (y/n): "
235			):
236			write_answer_file(actions)
237			print "Validating actions."
238			callback = CallbackClass()
239			actions, invalid_actions = validate_actions_result(actions, callback)
240			print "Generating test files."
241			generate_test_files(actions, invalid_actions)
242			print_invalid_actions(invalid_actions)
243			print "Test files have been generated."
244
245
246			if __name__ == "__main__":
247			main()
248

StackStorm / st2contrib

Pull Request — master (#543)

request_values() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like