view release on metacpan or  search on metacpan

src/subversion/build/run_tests.py  view on Meta::CPAN

#!/usr/bin/env python
#
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you 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.
#
#
#
# run_tests.py - run the tests in the regression test suite.
#

'''usage: python run_tests.py
            [--verbose] [--log-to-stdout] [--cleanup] [--parallel]
            [--url=<base-url>] [--http-library=<http-library>] [--enable-sasl]
            [--fs-type=<fs-type>] [--fsfs-packing] [--fsfs-sharding=<n>]
            [--list] [--milestone-filter=<regex>] [--mode-filter=<type>]
            [--server-minor-version=<version>] [--http-proxy=<host>:<port>]
            [--config-file=<file>] [--ssl-cert=<file>]
            <abs_srcdir> <abs_builddir>
            <prog ...>

The optional flags and the first two parameters are passed unchanged
to the TestHarness constructor.  All other parameters are names of
test programs.

Each <prog> should be the full path (absolute or from the current directory)
and filename of a test program, optionally followed by '#' and a comma-
separated list of test numbers; the default is to run all the tests in it.
'''

# A few useful constants
SVN_VER_MINOR = 8

import os, re, subprocess, sys, imp, threading
from datetime import datetime

import getopt
try:
  my_getopt = getopt.gnu_getopt
except AttributeError:
  my_getopt = getopt.getopt

# Ensure the compiled C tests use a known locale (Python tests set the locale
# explicitly).
os.environ['LC_ALL'] = 'C'

class TextColors:
  '''Some ANSI terminal constants for output color'''
  ENDC = '\033[0;m'
  FAILURE = '\033[1;31m'
  SUCCESS = '\033[1;32m'

  @classmethod
  def disable(cls):
    cls.ENDC = ''
    cls.FAILURE = ''
    cls.SUCCESS = ''


def _get_term_width():
  'Attempt to discern the width of the terminal'
  # This may not work on all platforms, in which case the default of 80
  # characters is used.  Improvements welcomed.

  def ioctl_GWINSZ(fd):
    try:
      import fcntl, termios, struct, os
      cr = struct.unpack('hh', fcntl.ioctl(fd, termios.TIOCGWINSZ, '1234'))
    except:
      return None
    return cr

  cr = None
  if not cr:
    try:
      cr = (os.environ['SVN_MAKE_CHECK_LINES'],
            os.environ['SVN_MAKE_CHECK_COLUMNS'])
    except:
      cr = None
  if not cr:
    cr = ioctl_GWINSZ(0) or ioctl_GWINSZ(1) or ioctl_GWINSZ(2)
  if not cr:
    try:
      fd = os.open(os.ctermid(), os.O_RDONLY)
      cr = ioctl_GWINSZ(fd)
      os.close(fd)
    except:
      pass
  if not cr:
    try:
      cr = (os.environ['LINES'], os.environ['COLUMNS'])
    except:
      cr = None
  if not cr:

src/subversion/build/run_tests.py  view on Meta::CPAN

      if not self.log or self.dots_written >= dot_count:
        return
      dots = (completed * dot_count) / total
      if dots > dot_count:
        dots = dot_count
      dots_to_write = dots - self.dots_written
      os.write(sys.stdout.fileno(), '.' * dots_to_write)
      self.dots_written = dots

    tests_completed = 0
    prog = subprocess.Popen(cmdline, stdout=subprocess.PIPE,
                            stderr=self.log)
    line = prog.stdout.readline()
    while line:
      if sys.platform == 'win32':
        # Remove CRs inserted because we parse the output as binary.
        line = line.replace('\r', '')

      # If using --log-to-stdout self.log in None.
      if self.log:
        self.log.write(line)

      if line.startswith('PASS') or line.startswith('FAIL') \
           or line.startswith('XFAIL') or line.startswith('XPASS') \
           or line.startswith('SKIP'):
        tests_completed += 1
        progress_func(tests_completed)

      line = prog.stdout.readline()

    # If we didn't run any tests, still print out the dots
    if not tests_completed:
      os.write(sys.stdout.fileno(), '.' * dot_count)

    prog.wait()
    return prog.returncode

  def _run_py_test(self, prog, test_nums, dot_count):
    'Run a python test, passing parameters as needed.'
    progdir, progbase = os.path.split(prog)

    old_path = sys.path[:]
    sys.path = [progdir] + sys.path

    try:
      prog_mod = imp.load_module(progbase[:-3], open(prog, 'r'), prog,
                                 ('.py', 'U', imp.PY_SOURCE))
    except:
      print("Don't know what to do about " + progbase)
      sys.exit(1)

    import svntest.main

    # set up our options
    svntest.main.create_default_options()
    if self.base_url is not None:
      svntest.main.options.test_area_url = self.base_url
    if self.enable_sasl is not None:
      svntest.main.options.enable_sasl = True
    if self.parallel is not None:
      svntest.main.options.parallel = svntest.main.default_num_threads
    if self.config_file is not None:
      svntest.main.options.config_file = self.config_file
    if self.verbose is not None:
      svntest.main.options.verbose = True
    if self.cleanup is not None:
      svntest.main.options.cleanup = True
    if self.fs_type is not None:
      svntest.main.options.fs_type = self.fs_type
    if self.http_library is not None:
      svntest.main.options.http_library = self.http_library
    if self.server_minor_version is not None:
      svntest.main.options.server_minor_version = int(self.server_minor_version)
    if self.list_tests is not None:
      svntest.main.options.list_tests = True
    if self.milestone_filter is not None:
      svntest.main.options.milestone_filter = self.milestone_filter
    if self.set_log_level is not None:
      # Somehow the logger is not setup correctly from win-tests.py, so
      # setting the log level would fail. ### Please fix
      if svntest.main.logger is None:
        import logging
        svntest.main.logger = logging.getLogger()
      svntest.main.logger.setLevel(self.set_log_level)
    if self.svn_bin is not None:
      svntest.main.options.svn_bin = self.svn_bin
    if self.fsfs_sharding is not None:
      svntest.main.options.fsfs_sharding = int(self.fsfs_sharding)
    if self.fsfs_packing is not None:
      svntest.main.options.fsfs_packing = self.fsfs_packing
    if self.mode_filter is not None:
      svntest.main.options.mode_filter = self.mode_filter
    if self.ssl_cert is not None:
      svntest.main.options.ssl_cert = self.ssl_cert
    if self.http_proxy is not None:
      svntest.main.options.http_proxy = self.http_proxy

    svntest.main.options.srcdir = self.srcdir

    # setup the output pipes
    if self.log:
      sys.stdout.flush()
      sys.stderr.flush()
      self.log.flush()
      old_stdout = os.dup(sys.stdout.fileno())
      old_stderr = os.dup(sys.stderr.fileno())
      os.dup2(self.log.fileno(), sys.stdout.fileno())
      os.dup2(self.log.fileno(), sys.stderr.fileno())

    # These have to be class-scoped for use in the progress_func()
    self.dots_written = 0
    self.progress_lock = threading.Lock()
    def progress_func(completed, total):
      """Report test suite progress. Can be called from multiple threads
         in parallel mode."""
      if not self.log:
        return
      dots = (completed * dot_count) / total
      if dots > dot_count:
        dots = dot_count
      self.progress_lock.acquire()
      if self.dots_written < dot_count:
        dots_to_write = dots - self.dots_written
        self.dots_written = dots
        os.write(old_stdout, '.' * dots_to_write)
      self.progress_lock.release()

    serial_only = hasattr(prog_mod, 'serial_only') and prog_mod.serial_only

    # run the tests
    svntest.testcase.TextColors.disable()

    if self.list_tests:
      prog_f = None
    else:
      prog_f = progress_func

    if test_nums:
      test_selection = [test_nums]
    else:
      test_selection = []

    try:
      failed = svntest.main.execute_tests(prog_mod.test_list,
                                          serial_only=serial_only,
                                          test_name=progbase,
                                          progress_func=prog_f,
                                          test_selection=test_selection)
    except svntest.Failure:
      if self.log:
        os.write(old_stdout, '.' * dot_count)
      failed = True

    # restore some values
    sys.path = old_path
    if self.log:
      sys.stdout.flush()
      sys.stderr.flush()
      os.dup2(old_stdout, sys.stdout.fileno())
      os.dup2(old_stderr, sys.stderr.fileno())
      os.close(old_stdout)
      os.close(old_stderr)

    return failed

  def _run_test(self, prog, test_nr, total_tests):
    "Run a single test. Return the test's exit code."

    if self.log:
      log = self.log
    else:
      log = sys.stdout

    test_nums = None