| #!/usr/bin/env python |
| # |
| # Copyright (c) 2016, The OpenThread Authors. |
| # All rights reserved. |
| # |
| # Redistribution and use in source and binary forms, with or without |
| # modification, are permitted provided that the following conditions are met: |
| # 1. Redistributions of source code must retain the above copyright |
| # notice, this list of conditions and the following disclaimer. |
| # 2. Redistributions in binary form must reproduce the above copyright |
| # notice, this list of conditions and the following disclaimer in the |
| # documentation and/or other materials provided with the distribution. |
| # 3. Neither the name of the copyright holder nor the |
| # names of its contributors may be used to endorse or promote products |
| # derived from this software without specific prior written permission. |
| # |
| # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" |
| # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE |
| # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE |
| # ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE |
| # LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR |
| # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF |
| # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS |
| # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN |
| # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) |
| # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE |
| # POSSIBILITY OF SUCH DAMAGE. |
| # |
| |
| |
| import argparse |
| import fnmatch |
| import logging |
| import json |
| import os |
| import sys |
| import time |
| import unittest |
| from collections import OrderedDict |
| |
| from autothreadharness.harness_case import HarnessCase |
| from autothreadharness.open_thread_controller import OpenThreadController |
| from autothreadharness import settings |
| |
| logging.basicConfig(level=logging.INFO) |
| |
| logger = logging.getLogger() |
| """Logger: The global logger""" |
| |
| logger.setLevel(logging.INFO) |
| |
| RESUME_SCRIPT_PATH = '%appdata%\\Microsoft\\Windows\\Start Menu\\Programs\\' \ |
| 'Startup\\continue_harness.bat' |
| |
| class SimpleTestResult(unittest.TestResult): |
| def __init__(self, path, auto_reboot_args=None): |
| """Record test results in json file |
| |
| Args: |
| path (str): File path to record the results |
| auto_reboot (bool): Whether reboot when harness die |
| """ |
| super(SimpleTestResult, self).__init__() |
| self.path = path |
| self.auto_reboot_args = auto_reboot_args |
| self.result = json.load(open(self.path, 'r')) |
| self.log_handler = None |
| self.started = None |
| logger.info('Initial state is %s', json.dumps(self.result, indent=2)) |
| |
| def startTest(self, test): |
| logger.info('\n========================================\n%s\n========================================', test.__class__.__name__) |
| |
| # create start up script if auto reboot enabled |
| if self.auto_reboot_args: |
| test.auto_reboot = True |
| os.system('echo %s > "%s"' |
| % (' '.join(self.auto_reboot_args + ['-c', test.__class__.__name__]), |
| RESUME_SCRIPT_PATH)) |
| |
| # record start timestamp |
| self.started = time.strftime('%Y-%m-%dT%H:%M:%S') |
| |
| os.system('mkdir %s' % test.result_dir) |
| self.log_handler = logging.FileHandler('%s\\auto-%s.log' % (test.result_dir, time.strftime('%Y%m%d%H%M%S'))) |
| self.log_handler.setLevel(logging.DEBUG) |
| self.log_handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(message)s')) |
| logger.addHandler(self.log_handler) |
| |
| def add_result(self, test, passed, error=None): |
| """Record test result into json file |
| |
| Args: |
| test (TestCase): The test just run |
| passed (bool): Whether the case is passed |
| """ |
| self.result[unicode(test.__class__.__name__)] = { |
| 'started': self.started, |
| 'stopped': time.strftime('%Y-%m-%dT%H:%M:%S'), |
| 'passed': passed, |
| 'error': error |
| } |
| if self.auto_reboot_args: |
| os.system('del "%s"' % RESUME_SCRIPT_PATH) |
| |
| json.dump(OrderedDict(sorted(self.result.items(), key=lambda t: t[0])), |
| open(self.path, 'w'), indent=2) |
| |
| # save logs |
| logger.removeHandler(self.log_handler) |
| self.log_handler.close() |
| self.log_handler = None |
| time.sleep(2) |
| |
| # close explorers |
| os.system('taskkill /f /im explorer.exe && start explorer.exe') |
| |
| def addSuccess(self, test): |
| logger.info('case[%s] pass', test.__class__.__name__) |
| super(SimpleTestResult, self).addSuccess(test) |
| self.add_result(test, True) |
| |
| def addFailure(self, test, err): |
| logger.warning('case[%s] fail', test.__class__.__name__) |
| super(SimpleTestResult, self).addFailure(test, err) |
| self.add_result(test, False) |
| |
| def addError(self, test, err): |
| logger.error('case[%s] error', test.__class__.__name__, exc_info=err) |
| |
| if err and err[0] is SystemExit: |
| if self.auto_reboot_args: |
| logger.warning('rebooting..') |
| os.system('shutdown /r /t 1') |
| else: |
| logger.warning('exiting..') |
| sys.exit(1) |
| |
| super(SimpleTestResult, self).addError(test, err) |
| self.add_result(test, None, str(err[1])) |
| |
| def list_devices(names=None, continue_from=None, **kwargs): |
| """List devices in settings file and print versions""" |
| |
| if not names: |
| names = [device for device, _type in settings.GOLDEN_DEVICES if _type == 'OpenThread'] |
| |
| if continue_from: |
| continue_from = names.index(continue_from) |
| else: |
| continue_from = 0 |
| |
| for port in names[continue_from:]: |
| try: |
| with OpenThreadController(port) as otc: |
| print('%s: %s' % (port, otc.version)) |
| except: |
| logger.exception('failed to get version of %s' % port) |
| |
| def discover(names=None, pattern=['*.py'], skip='efp', dry_run=False, blacklist=None, name_greps=None, |
| manual_reset=False, delete_history=False, max_devices=0, |
| continue_from=None, result_file='./result.json', auto_reboot=False): |
| """Discover all test cases and skip those passed |
| |
| Args: |
| pattern (str): Pattern to match case modules, refer python's unittest |
| documentation for more details |
| skip (str): types cases to skip |
| """ |
| if not os.path.exists(settings.OUTPUT_PATH): |
| os.mkdir(settings.OUTPUT_PATH) |
| |
| if delete_history: |
| os.system('del history.json') |
| |
| if blacklist: |
| try: |
| excludes = [line.strip('\n') for line in open(blacklist, 'r').readlines() |
| if not line.startswith('#')] |
| except: |
| logger.exception('Failed to open test case black list file') |
| raise |
| else: |
| excludes = [] |
| |
| log = None |
| if os.path.isfile(result_file): |
| try: |
| log = json.load(open(result_file, 'r')) |
| except: |
| logger.exception('Failed to open result file') |
| |
| if not log: |
| log = {} |
| json.dump(log, open(result_file, 'w'), indent=2) |
| |
| suite = unittest.TestSuite() |
| discovered = unittest.defaultTestLoader.discover('cases', pattern) |
| |
| if names and continue_from: |
| names = names[names.index(continue_from):] |
| |
| for s1 in discovered: |
| for s2 in s1: |
| for case in s2: |
| if case.__class__ is HarnessCase: |
| continue |
| case_name = unicode(case.__class__.__name__) |
| |
| # grep name |
| if name_greps and not any(fnmatch.fnmatch(case_name, name_grep) for name_grep in name_greps): |
| logger.info('case[%s] skipped by name greps', case_name) |
| continue |
| |
| # whitelist |
| if len(names) and case_name not in names: |
| logger.info('case[%s] skipped', case_name) |
| continue |
| |
| # skip cases |
| if case_name in log.keys(): |
| if (log[case_name]['passed'] and ('p' in skip)) \ |
| or (log[case_name]['passed'] is False and ('f' in skip)) \ |
| or (log[case_name]['passed'] is None and ('e' in skip)): |
| logger.warning('case[%s] skipped for its status[%s]', case_name, log[case_name]['passed']) |
| continue |
| |
| # continue from |
| if continue_from: |
| if continue_from != case_name: |
| logger.warning('case[%s] skipped for continue from[%s]', case_name, continue_from) |
| continue |
| else: |
| continue_from = None |
| |
| # black list |
| if case_name in excludes: |
| logger.warning('case[%s] skipped for blacklist', case_name) |
| continue |
| |
| # max devices |
| if max_devices and case.golden_devices_required > max_devices: |
| logger.warning('case[%s] skipped for exceeding max golden devices allowed[%d]', case_name, max_devices) |
| continue |
| |
| suite.addTest(case) |
| logger.info('case[%s] added', case_name) |
| |
| if auto_reboot: |
| argv = [] |
| argv.append('"%s"' % os.sep.join([os.getcwd(), 'start.bat'])) |
| argv.extend(['-p', pattern]) |
| argv.extend(['-k', skip]) |
| argv.extend(['-o', result_file]) |
| argv.append('-a') |
| |
| if manual_reset: |
| argv.append('-m') |
| |
| if delete_history: |
| argv.append('-d') |
| |
| auto_reboot_args = argv + names |
| else: |
| auto_reboot_args = None |
| os.system('del "%s"' % RESUME_SCRIPT_PATH) |
| |
| # manual reset |
| if manual_reset: |
| settings.PDU_CONTROLLER_TYPE = 'MANUAL_PDU_CONTROLLER' |
| settings.PDU_CONTROLLER_OPEN_PARAMS = {} |
| settings.PDU_CONTROLLER_REBOOT_PARAMS = {} |
| |
| result = SimpleTestResult(result_file, auto_reboot_args) |
| for case in suite: |
| logger.info(case.__class__.__name__) |
| |
| if dry_run: |
| return |
| |
| suite.run(result) |
| |
| def main(): |
| parser = argparse.ArgumentParser(description='Thread harness test case runner') |
| parser.add_argument('--auto-reboot', '-a', action='store_true', default=False, |
| help='restart system when harness service die') |
| parser.add_argument('names', metavar='NAME', type=str, nargs='*', default=None, |
| help='test case name, omit to test all') |
| parser.add_argument('--blacklist', '-b', metavar='BLACKLIST_FILE', type=str, |
| help='file to list test cases to skip', default=None) |
| parser.add_argument('--continue-from', '-c', type=str, default=None, |
| help='first case to test') |
| parser.add_argument('--delete-history', '-d', action='store_true', default=False, |
| help='clear history on startup') |
| parser.add_argument('--name-greps', '-g', action='append', default=None, |
| help='grep case by names') |
| parser.add_argument('--list-file', '-i', type=str, default=None, |
| help='file to list cases names to test') |
| parser.add_argument('--skip', '-k', metavar='SKIP', type=str, |
| help='type of results to skip.' \ |
| 'e for error, f for fail, p for pass.', default='') |
| parser.add_argument('--list-devices', '-l', action='store_true', default=False, |
| help='list devices') |
| parser.add_argument('--manual-reset', '-m', action='store_true', default=False, |
| help='reset devices manually') |
| parser.add_argument('--dry-run', '-n', action='store_true', default=False, |
| help='just show what to run') |
| parser.add_argument('--result-file', '-o', type=str, default=settings.OUTPUT_PATH + '\\result.json', |
| help='file to store and read current status') |
| parser.add_argument('--pattern', '-p', metavar='PATTERN', type=str, |
| help='file name pattern, default to "*.py"', default='*.py') |
| parser.add_argument('--max-devices', '-u', type=int, default=0, |
| help='max golden devices allowed') |
| |
| args = vars(parser.parse_args()) |
| |
| if args['list_file']: |
| try: |
| names = [line.strip('\n') for line in open(args['list_file'], 'r').readlines() |
| if not line.startswith('#')] |
| except: |
| logger.exception('Failed to open test case list file') |
| raise |
| else: |
| args['names'] = args['names'] + names |
| |
| args.pop('list_file') |
| |
| if args.pop('list_devices', False): |
| list_devices(**args) |
| else: |
| discover(**args) |
| |
| if __name__ == '__main__': |
| main() |
