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()