Report: data
[csit.git] / main.py
1 #!/usr/bin/env python
2
3 # Copyright (c) 2016 Cisco and/or its affiliates.
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # 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 """This is a helper script to make test execution easy."""
17
18 from __future__ import print_function
19 import sys
20 import os
21 import time
22 from string import ascii_lowercase
23 from random import sample
24 import argparse
25 from pykwalify.core import Core
26 from pykwalify.errors import PyKwalifyException
27 from yaml import load
28 import robot
29 from robot.errors import DATA_ERROR, DataError, FRAMEWORK_ERROR, FrameworkError
30 from robot.run import RobotFramework
31 from robot.conf.settings import RobotSettings
32 from robot.running.builder import TestSuiteBuilder
33 from robot.running.model import TestSuite
34
35 TOPOLOGIES_DIR = './topologies/enabled/'
36 TESTS_DIR = './tests'
37 OUTPUTS_DIR = './outputs'
38
39
40 def get_suite_list(*datasources, **options):
41     """Returns filtered test suites based on include exclude tags
42
43     :param datasources: paths to tests
44     :param options: Robot Framework options (robot.conf.settings.py)
45     :return: list of Robot Framework TestSuites which contain tests
46     """
47     class _MyRobotFramework(RobotFramework):
48         """Custom implementation of RobotFramework main()."""
49         def main(self, datasources, **options):
50             # copied from robot.run.RobotFramework.main
51             settings = RobotSettings(options)
52             test_suite = TestSuiteBuilder(settings['SuiteNames'],
53                                           settings['WarnOnSkipped'])
54             # pylint: disable=star-args
55             suite = test_suite.build(*datasources)
56             suite.configure(**settings.suite_config)
57
58             return suite
59
60     # get all test cases list without run tests, execute runs overloaded main
61     # function
62     suite = _MyRobotFramework().execute(*datasources, output=None, dryrun=True,
63                                         **options)
64     if isinstance(suite, TestSuite):
65         suites = []
66         suites.append(suite)
67         append_new = True
68         while append_new:
69             append_new = False
70             tmp = []
71             for suite in suites:
72                 # pylint: disable=protected-access
73                 if len(suite.suites._items) > 0:
74                     for i in suite.suites._items:
75                         tmp.append(i)
76                     append_new = True
77                 else:
78                     tmp.append(suite)
79             suites = tmp
80         return suites
81         # TODO: check testcases Tags ? all tests should have same set of tags
82     else:
83         if suite == DATA_ERROR:
84             raise DataError
85         if suite == FRAMEWORK_ERROR:
86             raise FrameworkError
87         return []
88
89
90 def run_suites(tests_dir, suites, output_dir, output_prefix='suite',
91                **options):
92     """Execute RF's run with parameters."""
93
94     with open('{}/{}.out'.format(output_dir, output_prefix), 'w') as out:
95         robot.run(tests_dir,
96                   suite=[s.longname for s in suites],
97                   output='{}/{}.xml'.format(output_dir, output_prefix),
98                   debugfile='{}/{}.log'.format(output_dir, output_prefix),
99                   log=None,
100                   report=None,
101                   stdout=out,
102                   **options)
103
104
105 def parse_outputs(output_dir):
106     """Parse output xmls from all executed tests."""
107
108     outs = [os.path.join(output_dir, file_name)
109             for file_name in os.listdir(output_dir)
110             if file_name.endswith('.xml')]
111     # pylint: disable=star-args
112     robot.rebot(*outs, merge=True)
113
114
115 def topology_lookup(topology_paths, topo_dir, validate):
116     """Make topology list and validate topologies against schema
117
118     :param parsed_args: topology list, is empty then scans topologies in
119                         topo_dir
120     :param topo_dir: scan directory for topologies
121     :param validate: if True then validate topology
122     :return: list of topologies
123     """
124
125     ret_topologies = []
126     if topology_paths:
127         for topo in topology_paths:
128             if os.path.exists(topo):
129                 ret_topologies.append(topo)
130             else:
131                 print("Topology file {} doesn't exist".format(topo),
132                       file=sys.stderr)
133     else:
134         ret_topologies = [os.path.join(topo_dir, file_name)
135                           for file_name in os.listdir(topo_dir)
136                           if file_name.lower().endswith('.yaml')]
137
138     if len(ret_topologies) == 0:
139         print('No valid topology found', file=sys.stderr)
140         exit(1)
141
142     # validate topologies against schema
143     exit_on_error = False
144     for topology_name in ret_topologies:
145         try:
146             with open(topology_name) as file_name:
147                 yaml_obj = load(file_name)
148             core = Core(source_file=topology_name,
149                         schema_files=yaml_obj["metadata"]["schema"])
150             core.validate()
151         except PyKwalifyException as ex:
152             print('Unable to verify topology {}, schema error: {}'.\
153                   format(topology_name, ex),
154                   file=sys.stderr)
155             exit_on_error = True
156         except KeyError as ex:
157             print('Unable to verify topology {}, key error: {}'.\
158                   format(topology_name, ex),
159                   file=sys.stderr)
160             exit_on_error = True
161         except Exception as ex:
162             print('Unable to verify topology {}, {}'.format(topology_name, ex),
163                   file=sys.stderr)
164             exit_on_error = True
165
166     if exit_on_error and validate:
167         exit(1)
168
169     return ret_topologies
170
171
172 def main():
173     """Main function."""
174     parser = argparse.ArgumentParser(description='A test runner')
175     parser.add_argument('-i', '--include', action='append',
176                         help='include tests with tag')
177     parser.add_argument('-e', '--exclude', action='append',
178                         help='exclude tests with tag')
179     parser.add_argument('-s', '--suite', action='append',
180                         help='full name of suite to run')
181     parser.add_argument('-t', '--topology', action='append',
182                         help='topology where tests should be run')
183     parser.add_argument('-d', '--test_dir', nargs='?', default=TESTS_DIR,
184                         help='where tests are stored')
185     parser.add_argument('-o', '--output_dir', nargs='?', default=OUTPUTS_DIR,
186                         help='where results are stored')
187     parser.add_argument('-L', '--loglevel', nargs='?', default='INFO', type=str,
188                         choices=['TRACE', 'DEBUG', 'INFO', 'WARN', 'NONE'],
189                         help='robot frameworks level for logging')
190     parser.add_argument('-n', '--no_validate', action="store_false",
191                         help='Do not exit if topology validation failed')
192
193     args = parser.parse_args()
194
195     i = args.include or []
196     excl = args.exclude or []
197     suite_filter = args.suite or []
198     test_dir = args.test_dir
199
200     # prepare output subdir
201     suite_output_dir = os.path.join(args.output_dir,
202                                     time.strftime('%y%m%d%H%M%S'))
203     os.makedirs(suite_output_dir)
204
205     topologies = topology_lookup(args.topology, TOPOLOGIES_DIR,
206                                  args.no_validate)
207     suite_list = get_suite_list(test_dir, include=i, exclude=excl,
208                                 suite=suite_filter)
209
210     # TODO: do the topology suite mapping magic
211     #       for now all tests on single topology
212     if len(topologies) > 1:
213         print('Multiple topologies unsupported yet', file=sys.stderr)
214         exit(1)
215     topology_suite_mapping = {topologies[0]: suite_list}
216
217     # on all topologies, run test
218     # TODO: run parallel
219     for topology_path, topology_suite_list in topology_suite_mapping.items():
220         topology_path_variable = 'TOPOLOGY_PATH:{}'.format(topology_path)
221         variables = [topology_path_variable]
222         print('Runing tests on topology {}'.format(topology_path))
223         run_suites(test_dir, topology_suite_list, variable=variables,
224                    output_dir=suite_output_dir,
225                    output_prefix=''.join(sample(ascii_lowercase, 5)),
226                    include=i, exclude=excl, loglevel=args.loglevel)
227
228     print('Parsing test results')
229     parse_outputs(suite_output_dir)
230
231
232 if __name__ == "__main__":
233     main()

©2016 FD.io a Linux Foundation Collaborative Project. All Rights Reserved.
Linux Foundation is a registered trademark of The Linux Foundation. Linux is a registered trademark of Linus Torvalds.
Please see our privacy policy and terms of use.