run_tests.py

#!/bin/env python
# 

# Copyright (c) 2012-2017 The Khronos Group Inc.
#
# Licensed 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.
#


import os
import sys
import traceback

import datetime
import time

import subprocess
import threading

import re

from pprint import pprint

class BackgroundProcess(threading.Thread):
    def __init__(self, **args):
        self.args = args
        self.stdout = None
        self.stderr = None
        self.process = None
        threading.Thread.__init__(self)

    def run(self):
        self.process = subprocess.Popen(
            stderr=subprocess.PIPE,
            **self.args)
        self.stdout, self.stderr = self.process.communicate()

report_re = re.compile(
        r'#REPORT:'
        r' (?P<timestamp>.{14})'
        r' (?P<testid>.*)'
        r' (?P<total>\d+)'
        r' (?P<disabled>\d+)'
        r' (?P<started>\d+)'
        r' (?P<completed>\d+)'
        r' (?P<passed>\d+)'
        r' (?P<failed>\d+)'
        r' (?P<version>[^$]+)'
)

class TestRunner(object):

    tests = []

    timeout = float(os.environ.get('VX_TEST_TIMEOUT', '65'))  # seconds

    testid = '?'
    total_tests = 0
    total_disabled_tests = 0
    total_started_tests = 0
    total_completed_tests = 0
    total_passed_tests = 0
    total_failed_tests = 0
    tests_version = 'unknown'

    def get_test_list(self):
        p = subprocess.Popen(
                             args=self.launch_args + ['--quiet', '--list_tests', '--run_disabled'],
                             stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        stdout, stderr = p.communicate()
        if p.returncode != 0:
            raise Exception("Can't get list of tests")

        lines = stdout.replace('\r\n', '\n').split('\n');
        for line in lines:
            if line == "":
                continue
            self.tests.append(line)

        m = re.search(report_re, stderr)
        if m:
            self.testid = m.group('testid')

    def run_test(self, test):
        self.total_started_tests += 1

        bp = BackgroundProcess(
                args=self.launch_args + ['--quiet', '--filter=%s' % test.replace(':', '*')]
                )
        bp.start()

        bp.join(self.timeout)
        if bp.is_alive():
            print('#TIMEOUT on test "%s". TERMINATING' % test)
            sys.stdout.flush()
            try:
                bp.process.terminate()
            except:
                pass
            bp.join()

        m = re.search(report_re, bp.stderr)
        if m:
            timestamp = m.group('timestamp')
            testid = m.group('testid')
            total = m.group('total')
            disabled = m.group('disabled')
            started = m.group('started')
            completed = m.group('completed')
            passed = m.group('passed')
            failed = m.group('failed')
            if m.group('version'):
                self.tests_version = m.group('version')
            if str(started) == '0' and str(disabled) == '0':
                print("#CHECK FILTER: %s" % test)
                self.total_failed_tests += 1
            if str(disabled) != '0':
                 self.total_disabled_tests += 1
                 self.total_started_tests -= 1
            if str(completed) != '0':
                 self.total_completed_tests += 1
            if str(passed) != '0':
                 self.total_passed_tests += 1
            if str(failed) != '0':
                 self.total_failed_tests += 1
        else:
            self.total_failed_tests += 1
            sys.stdout.write(bp.stderr)
            if bp.process.returncode != 0:
                print('Process exit code: %d' % bp.process.returncode)

    def printUsage(self):
        print('''\
Usage:
    run_tests.py <vx_test_conformance executable> <filter and other parameters>

Environment variables:
    VX_TEST_DATA_PATH - path to test_data directory (used by vx_test_conformance)
    VX_TEST_TIMEOUT - single test timeout (in seconds)

Example:
    run_tests.py ./bin/vx_test_conformance
    run_tests.py ./bin/vx_test_conformance --filter=*Canny*\
''')

    def run(self):
        try:
            if len(sys.argv) < 2:
                print("Missed executable path")
                self.printUsage()
                return 2
            if sys.argv[1] in ['-h', '--help', '/?']:
                self.printUsage()
                return 0

            self.launch_args = sys.argv[1:]

            self.get_test_list()

            self.launch_args = [a for a in self.launch_args if not a.startswith('--filter=')]

            self.total_tests = len(self.tests)

            print('#FOUND %d tests' % self.total_tests)
            print('Test timeout=%s' % self.timeout)
            print('')
            sys.stdout.flush()

            prev = 0
            i = 0
            for t in self.tests:
                if (self.total_tests >= 500):
                    next = i * 100 / self.total_tests
                    if int(next) != prev:
                        print('# %02d%%' % next)
                        prev = next
                i += 1
                sys.stdout.flush()
                sys.stderr.flush()
                try:
                    self.run_test(t)
                except KeyboardInterrupt:
                    break
                except:
                    print traceback.format_exc()

            print('')
            print('ALL DONE')

            timestamp = datetime.datetime.now().strftime('%Y%m%d%H%M%S')

            print('')
            print('#REPORT: %s %s %d %d %d %d %d %d %s' % (
                    timestamp, self.testid,
                    self.total_tests,
                    self.total_disabled_tests,
                    self.total_started_tests,
                    self.total_completed_tests,
                    self.total_passed_tests,
                    self.total_failed_tests,
                    self.tests_version))

            return 0 if (self.total_tests == (self.total_started_tests + self.total_disabled_tests) and self.total_failed_tests == 0) else 1
        except:
            print traceback.format_exc()

if __name__ == "__main__":
    sys.exit(TestRunner().run())