<commit_before>#
#  main.py: a shared, automated test suite for Subversion
#
#  Subversion is a tool for revision control.
#  See http://subversion.tigris.org for more information.
#
# ====================================================================
#    Licensed to the Subversion Corporation (SVN Corp.) under one
#    or more contributor license agreements.  See the NOTICE file
#    distributed with this work for additional information
#    regarding copyright ownership.  The SVN Corp. 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.
######################################################################

import sys     # for argv[]
import os
import shutil  # for rmtree()
import re
import stat    # for ST_MODE
import subprocess
import copy    # for deepcopy()
import time    # for time()
import traceback # for print_exc()
import threading
try:
  # Python >=3.0
  import queue
  from urllib.parse import quote as urllib_parse_quote
  from urllib.parse import unquote as urllib_parse_unquote
except ImportError:
  # Python <3.0
  import Queue as queue
  from urllib import quote as urllib_parse_quote
  from urllib import unquote as urllib_parse_unquote

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

import svntest
from svntest import Failure
from svntest import Skip


######################################################################
#
#  HOW TO USE THIS MODULE:
#
#  Write a new python script that
#
#     1) imports this 'svntest' package
#
#     2) contains a number of related 'test' routines.  (Each test
#        routine should take no arguments, and return None on success
#        or throw a Failure exception on failure.  Each test should
#        also contain a short docstring.)
#
#     3) places all the tests into a list that begins with None.
#
#     4) calls svntest.main.client_test() on the list.
#
#  Also, your tests will probably want to use some of the common
#  routines in the 'Utilities' section below.
#
#####################################################################
# Global stuff

class SVNProcessTerminatedBySignal(Failure):
  "Exception raised if a spawned process segfaulted, aborted, etc."
  pass

class SVNLineUnequal(Failure):
  "Exception raised if two lines are unequal"
  pass

class SVNUnmatchedError(Failure):
  "Exception raised if an expected error is not found"
  pass

class SVNCommitFailure(Failure):
  "Exception raised if a commit failed"
  pass

class SVNRepositoryCopyFailure(Failure):
  "Exception raised if unable to copy a repository"
  pass

class SVNRepositoryCreateFailure(Failure):
  "Exception raised if unable to create a repository"
  pass

# Windows specifics
if sys.platform == 'win32':
  windows = True
  file_scheme_prefix = 'file:///'
  _exe = '.exe'
  _bat = '.bat'
else:
  windows = False
  file_scheme_prefix = 'file://'
  _exe = ''
  _bat = ''

# The location of our mock svneditor script.
if windows:
  svneditor_script = os.path.join(sys.path[0], 'svneditor.bat')
else:
  svneditor_script = os.path.join(sys.path[0], 'svneditor.py')

# Username and password used by the working copies
wc_author = 'jrandom'
wc_passwd = '<PASSWORD>'

# Username and password used by the working copies for "second user"
# scenarios
wc_author2 = 'jconstant' # use the same password as wc_author

# Set C locale for command line programs
os.environ['LC_ALL'] = 'C'

# This function mimics the Python 2.3 urllib function of the same name.
def pathname2url(path):
  """Convert the pathname PATH from the local syntax for a path to the form
  used in the path component of a URL. This does not produce a complete URL.
  The return value will already be quoted using the quote() function."""
  return urllib_parse_quote(path.replace('\\', '/'))

# This function mimics the Python 2.3 urllib function of the same name.
def url2pathname(path):
  """Convert the path component PATH from an encoded URL to the local syntax
  for a path. This does not accept a complete URL. This function uses
  unquote() to decode PATH."""
  return os.path.normpath(urllib_parse_unquote(path))

######################################################################
# Global variables set during option parsing.  These should not be used
# until the variable command_line_parsed has been set to True, as is
# done in run_tests below.
command_line_parsed = False

# The locations of the svn, svnadmin and svnlook binaries, relative to
# the only scripts that import this file right now (they live in ../).
# Use --bin to override these defaults.
svn_binary = os.path.abspath('../../svn/svn' + _exe)
svnadmin_binary = os.path.abspath('../../svnadmin/svnadmin' + _exe)
svnlook_binary = os.path.abspath('../../svnlook/svnlook' + _exe)
svnsync_binary = os.path.abspath('../../svnsync/svnsync' + _exe)
svnversion_binary = os.path.abspath('../../svnversion/svnversion' + _exe)
svndumpfilter_binary = os.path.abspath('../../svndumpfilter/svndumpfilter' + \
                                       _exe)
entriesdump_binary = os.path.abspath('entries-dump' + _exe)

# Global variable indicating if we want verbose output, that is,
# details of what commands each test does as it does them.  This is
# incompatible with quiet_mode.
verbose_mode = False

# Global variable indicating if we want quiet output, that is, don't
# show PASS, XFAIL, or SKIP notices, but do show FAIL and XPASS.  This
# is incompatible with verbose_mode.
quiet_mode = False

# Global variable indicating if we want test data cleaned up after success
cleanup_mode = False

# Global variable indicating if svnserve should use Cyrus SASL
enable_sasl = False

# Global variable indicating that SVNKit binaries should be used
use_jsvn = False

# Global variable indicating which DAV library, if any, is in use
# ('neon', 'serf')
http_library = None

# Global variable: Number of shards to use in FSFS
# 'None' means "use FSFS's default"
fsfs_sharding = None

# Global variable: automatically pack FSFS repositories after every commit
fsfs_packing = None

# Configuration file (copied into FSFS fsfs.conf).
config_file = None

# Global variable indicating what the minor version of the server
# tested against is (4 for 1.4.x, for example).
server_minor_version = 7

# Global variable indicating if this is a child process and no cleanup
# of global directories is needed.
is_child_process = False

# Global URL to testing area.  Default to ra_local, current working dir.
test_area_url = file_scheme_prefix + pathname2url(os.path.abspath(os.getcwd()))

# Location to the pristine repository, will be calculated from test_area_url
# when we know what the user specified for --url.
pristine_url = None

# Global variable indicating the FS type for repository creations.
fs_type = None

# End of command-line-set global variables.
######################################################################

# All temporary repositories and working copies are created underneath
# this dir, so there's one point at which to mount, e.g., a ramdisk.
work_dir = "svn-test-work"

# Constant for the merge info property.
SVN_PROP_MERGEINFO = "svn:mergeinfo"

# Where we want all the repositories and working copies to live.
# Each test will have its own!
general_repo_dir = os.path.join(work_dir, "repositories")
general_wc_dir = os.path.join(work_dir, "working_copies")

# temp directory in which we will create our 'pristine' local
# repository and other scratch data.  This should be removed when we
# quit and when we startup.
temp_dir = os.path.join(work_dir, 'local_tmp')

# (derivatives of the tmp dir.)
pristine_dir = os.path.join(temp_dir, "repos")
greek_dump_dir = os.path.join(temp_dir, "greekfiles")
default_config_dir = os.path.abspath(os.path.join(temp_dir, "config"))

#
# Our pristine greek-tree state.
#
# If a test wishes to create an "expected" working-copy tree, it should
# call main.greek_state.copy().  That method will return a copy of this
# State object which can then be edited.
#
_item = svntest.wc.StateItem
greek_state = svntest.wc.State('', {
  'iota'        : _item("This is the file 'iota'.\n"),
  'A'           : _item(),
  'A/mu'        : _item("This is the file 'mu'.\n"),
  'A/B'         : _item(),
  'A/B/lambda'  : _item("This is the file 'lambda'.\n"),
  'A/B/E'       : _item(),
  'A/B/E/alpha' : _item("This is the file 'alpha'.\n"),
  'A/B/E/beta'  : _item("This is the file 'beta'.\n"),
  'A/B/F'       : _item(),
  'A/C'         : _item(),
  'A/D'         : _item(),
  'A/D/gamma'   : _item("This is the file 'gamma'.\n"),
  'A/D/G'       : _item(),
  'A/D/G/pi'    : _item("This is the file 'pi'.\n"),
  'A/D/G/rho'   : _item("This is the file 'rho'.\n"),
  'A/D/G/tau'   : _item("This is the file 'tau'.\n"),
  'A/D/H'       : _item(),
  'A/D/H/chi'   : _item("This is the file 'chi'.\n"),
  'A/D/H/psi'   : _item("This is the file 'psi'.\n"),
  'A/D/H/omega' : _item("This is the file 'omega'.\n"),
  })


######################################################################
# Utilities shared by the tests
def wrap_ex(func):
  "Wrap a function, catch, print and ignore exceptions"
  def w(*args, **kwds):
    try:
      return func(*args, **kwds)
    except Failure, ex:
      if ex.__class__ != Failure or ex.args:
        ex_args = str(ex)
        if ex_args:
          print('EXCEPTION: %s: %s' % (ex.__class__.__name__, ex_args))
        else:
          print('EXCEPTION: %s' % ex.__class__.__name__)
  return w

def setup_development_mode():
  "Wraps functions in module actions"
  l = [ 'run_and_verify_svn',
        'run_and_verify_svnversion',
        'run_and_verify_load',
        'run_and_verify_dump',
        'run_and_verify_checkout',
        'run_and_verify_export',
        'run_and_verify_update',
        'run_and_verify_merge',
        'run_and_verify_merge2',
        'run_and_verify_switch',
        'run_and_verify_commit',
        'run_and_verify_unquiet_status',
        'run_and_verify_status',
        'run_and_verify_diff_summarize',
        'run_and_verify_diff_summarize_xml',
        'run_and_validate_lock']

  for func in l:
    setattr(svntest.actions, func, wrap_ex(getattr(svntest.actions, func)))

def get_admin_name():
  "Return name of SVN administrative subdirectory."

  if (windows or sys.platform == 'cygwin') \
      and 'SVN_ASP_DOT_NET_HACK' in os.environ:
    return '_svn'
  else:
    return '.svn'

def get_start_commit_hook_path(repo_dir):
  "Return the path of the start-commit-hook conf file in REPO_DIR."

  return os.path.join(repo_dir, "hooks", "start-commit")


def get_pre_commit_hook_path(repo_dir):
  "Return the path of the pre-commit-hook conf file in REPO_DIR."

  return os.path.join(repo_dir, "hooks", "pre-commit")


def get_post_commit_hook_path(repo_dir):
  "Return the path of the post-commit-hook conf file in REPO_DIR."

  return os.path.join(repo_dir, "hooks", "post-commit")

def get_pre_revprop_change_hook_path(repo_dir):
  "Return the path of the pre-revprop-change hook script in REPO_DIR."

  return os.path.join(repo_dir, "hooks", "pre-revprop-change")

def get_svnserve_conf_file_path(repo_dir):
  "Return the path of the svnserve.conf file in REPO_DIR."

  return os.path.join(repo_dir, "conf", "svnserve.conf")

def get_fsfs_conf_file_path(repo_dir):
  "Return the path of the fsfs.conf file in REPO_DIR."

  return os.path.join(repo_dir, "db", "fsfs.conf")

def get_fsfs_format_file_path(repo_dir):
  "Return the path of the format file in REPO_DIR."

  return os.path.join(repo_dir, "db", "format")

# Run any binary, logging the command line and return code
def run_command(command, error_expected, binary_mode=0, *varargs):
  """Run COMMAND with VARARGS; return exit code as int; stdout, stderr
  as lists of lines.
  If ERROR_EXPECTED is None, any stderr also will be printed."""

  return run_command_stdin(command, error_expected, binary_mode,
                           None, *varargs)

# A regular expression that matches arguments that are trivially safe
# to pass on a command line without quoting on any supported operating
# system:
_safe_arg_re = re.compile(r'^[A-Za-z\d\.\_\/\-\:\@]+$')

def _quote_arg(arg):
  """Quote ARG for a command line.

  Simply surround every argument in double-quotes unless it contains
  only universally harmless characters.

  WARNING: This function cannot handle arbitrary command-line
  arguments.  It can easily be confused by shell metacharacters.  A
  perfect job would be difficult and OS-dependent (see, for example,
  http://msdn.microsoft.com/library/en-us/vccelng/htm/progs_12.asp).
  In other words, this function is just good enough for what we need
  here."""

  arg = str(arg)
  if _safe_arg_re.match(arg):
    return arg
  else:
    if os.name != 'nt':
      arg = arg.replace('$', '\$')
    return '"%s"' % (arg,)

def open_pipe(command, stdin=None, stdout=None, stderr=None):
  """Opens a subprocess.Popen pipe to COMMAND using STDIN,
  STDOUT, and STDERR.

  Returns (infile, outfile, errfile, waiter); waiter
  should be passed to wait_on_pipe."""
  command = [str(x) for x in command]

  # On Windows subprocess.Popen() won't accept a Python script as
  # a valid program to execute, rather it wants the Python executable.
  if (sys.platform == 'win32') and (command[0].endswith('.py')):
    command.insert(0, sys.executable)

  # Quote only the arguments on Windows.  Later versions of subprocess,
  # 2.5.2+ confirmed, don't require this quoting, but versions < 2.4.3 do.
  if (sys.platform == 'win32'):
    args = command[1:]
    args = ' '.join([_quote_arg(x) for x in args])
    command = command[0] + ' ' + args

  if not stdin:
    stdin = subprocess.PIPE
  if not stdout:
    stdout = subprocess.PIPE
  if not stderr:
    stderr = subprocess.PIPE

  p = subprocess.Popen(command,
                       stdin=stdin,
                       stdout=stdout,
                       stderr=stderr,
                       close_fds=not windows)
  return p.stdin, p.stdout, p.stderr, (p, command)

def wait_on_pipe(waiter, binary_mode, stdin=None):
  """Waits for KID (opened with open_pipe) to finish, dying
  if it does.  If kid fails create an error message containing
  any stdout and stderr from the kid.  Returns kid's exit code,
  stdout and stderr (the latter two as lists)."""
  if waiter is None:
    return

  kid, command = waiter
  stdout, stderr = kid.communicate(stdin)
  exit_code = kid.returncode

  # Normalize Windows line endings if in text mode.
  if windows and not binary_mode:
    stdout = stdout.replace('\r\n', '\n')
    stderr = stderr.replace('\r\n', '\n')

  # Convert output strings to lists.
  stdout_lines = stdout.splitlines(True)
  stderr_lines = stderr.splitlines(True)

  if exit_code < 0:
    exit_signal = os.WTERMSIG(-exit_code)
    if stdout_lines is not None:
      sys.stdout.write("".join(stdout_lines))
    if stderr_lines is not None:
      sys.stderr.write("".join(stderr_lines))
    if verbose_mode:
      # show the whole path to make it easier to start a debugger
      sys.stderr.write("CMD: %s terminated by signal %d\n"
                       % (' '.join(command), exit_signal))
    raise SVNProcessTerminatedBySignal
  else:
    if exit_code and verbose_mode:
      sys.stderr.write("CMD: %s exited with %d\n"
                       % (' '.join(command), exit_code))
    return stdout_lines, stderr_lines, exit_code

# Run any binary, supplying input text, logging the command line
def spawn_process(command, binary_mode=0, stdin_lines=None, *varargs):
  if stdin_lines and not isinstance(stdin_lines, list):
    raise TypeError("stdin_lines should have list type")

  # Log the command line
  if verbose_mode and not command.endswith('.py'):
    sys.stdout.write('CMD: %s %s\n' % (os.path.basename(command),
                                      ' '.join([_quote_arg(x) for x in varargs])))
    sys.stdout.flush()

  infile, outfile, errfile, kid = open_pipe([command] + list(varargs))

  if stdin_lines:
    for x in stdin_lines:
      infile.write(x)

  stdout_lines, stderr_lines, exit_code = wait_on_pipe(kid, binary_mode)
  infile.close()

  outfile.close()
  errfile.close()

  return exit_code, stdout_lines, stderr_lines

def run_command_stdin(command, error_expected, binary_mode=0,
                      stdin_lines=None, *varargs):
  """Run COMMAND with VARARGS; input STDIN_LINES (a list of strings
  which should include newline characters) to program via stdin - this
  should not be very large, as if the program outputs more than the OS
  is willing to buffer, this will deadlock, with both Python and
  COMMAND waiting to write to each other for ever.
  Return exit code as int; stdout, stderr as lists of lines.
  If ERROR_EXPECTED is None, any stderr also will be printed."""

  if verbose_mode:
    start = time.time()

  exit_code, stdout_lines, stderr_lines = spawn_process(command,
                                                        binary_mode,
                                                        stdin_lines,
                                                        *varargs)

  if verbose_mode:
    stop = time.time()
    print('<TIME = %.6f>' % (stop - start))
    for x in stdout_lines:
      sys.stdout.write(x)
    for x in stderr_lines:
      sys.stdout.write(x)

  if (not error_expected) and (stderr_lines):
    if not verbose_mode:
      for x in stderr_lines:
        sys.stdout.write(x)
    raise Failure

  return exit_code, \
         [line for line in stdout_lines if not line.startswith("DBG:")], \
         stderr_lines

def create_config_dir(cfgdir, config_contents=None, server_contents=None):
  "Create config directories and files"

  # config file names
  cfgfile_cfg = os.path.join(cfgdir, 'config')
  cfgfile_srv = os.path.join(cfgdir, 'servers')

  # create the directory
  if not os.path.isdir(cfgdir):
    os.makedirs(cfgdir)

  # define default config file contents if none provided
  if config_contents is None:
    config_contents = """
#
[auth]
password-stores =

[miscellany]
interactive-conflicts = false
"""

  # define default server file contents if none provided
  if server_contents is None:
    http_library_str = ""
    if http_library:
      http_library_str = "http-library=%s" % (http_library)
    server_contents = """
#
[global]
%s
store-plaintext-passwords=yes
store-passwords=yes
""" % (http_library_str)

  file_write(cfgfile_cfg, config_contents)
  file_write(cfgfile_srv, server_contents)

def _with_config_dir(args):
  if '--config-dir' in args:
    return args
  else:
    return args + ('--config-dir', default_config_dir)

def _with_auth(args):
  assert '--password' not in args
  args = args + ('--password', wc_passwd,
                 '--no-auth-cache' )
  if '--username' in args:
    return args
  else:
    return args + ('--username', wc_author )

# For running subversion and returning the output
def run_svn(error_expected, *varargs):
  """Run svn with VARARGS; return exit code as int; stdout, stderr as
  lists of lines.
  If ERROR_EXPECTED is None, any stderr also will be printed.  If
  you're just checking that something does/doesn't come out of
  stdout/stderr, you might want to use actions.run_and_verify_svn()."""
  return run_command(svn_binary, error_expected, 0,
                     *(_with_auth(_with_config_dir(varargs))))

# For running svnadmin.  Ignores the output.
def run_svnadmin(*varargs):
  """Run svnadmin with VARARGS, returns exit code as int; stdout, stderr as
  list of lines."""
  return run_command(svnadmin_binary, 1, 0, *varargs)

# For running svnlook.  Ignores the output.
def run_svnlook(*varargs):
  """Run svnlook with VARARGS, returns exit code as int; stdout, stderr as
  list of lines."""
  return run_command(svnlook_binary, 1, 0, *varargs)

def run_svnsync(*varargs):
  """Run svnsync with VARARGS, returns exit code as int; stdout, stderr as
  list of lines."""
  return run_command(svnsync_binary, 1, 0, *(_with_config_dir(varargs)))

def run_svnversion(*varargs):
  """Run svnversion with VARARGS, returns exit code as int; stdout, stderr
  as list of lines."""
  return run_command(svnversion_binary, 1, 0, *varargs)

def run_entriesdump(path):
  """Run the entries-dump helper, returning a dict of Entry objects."""
  # use spawn_process rather than run_command to avoid copying all the data
  # to stdout in verbose mode.
  exit_code, stdout_lines, stderr_lines = spawn_process(entriesdump_binary,
                                                        0, None, path)
  if verbose_mode:
    ### finish the CMD output
    print
  if exit_code or stderr_lines:
    ### report on this? or continue to just skip it?
    return None

  class Entry(object):
    pass
  entries = { }
  exec(''.join([line for line in stdout_lines if not line.startswith("DBG:")]))
  return entries


# Chmod recursively on a whole subtree
def chmod_tree(path, mode, mask):
  for dirpath, dirs, files in os.walk(path):
    for name in dirs + files:
      fullname = os.path.join(dirpath, name)
      if not os.path.islink(fullname):
        new_mode = (os.stat(fullname)[stat.ST_MODE] & ~mask) | mode
        os.chmod(fullname, new_mode)

# For clearing away working copies
def safe_rmtree(dirname, retry=0):
  "Remove the tree at DIRNAME, making it writable first"
  def rmtree(dirname):
    chmod_tree(dirname, 0666, 0666)
    shutil.rmtree(dirname)

  if not os.path.exists(dirname):
    return

  if retry:
    for delay in (0.5, 1, 2, 4):
      try:
        rmtree(dirname)
        break
      except:
        time.sleep(delay)
    else:
      rmtree(dirname)
  else:
    rmtree(dirname)

# For making local mods to files
def file_append(path, new_text):
  "Append NEW_TEXT to file at PATH"
  open(path, 'a').write(new_text)

# Append in binary mode
def file_append_binary(path, new_text):
  "Append NEW_TEXT to file at PATH in binary mode"
  open(path, 'ab').write(new_text)

# For creating new files, and making local mods to existing files.
def file_write(path, contents, mode='w'):
  """Write the CONTENTS to the file at PATH, opening file using MODE,
  which is (w)rite by default."""
  open(path, mode).write(contents)

# For reading the contents of a file
def file_read(path, mode = 'r'):
  """Return the contents of the file at PATH, opening file using MODE,
  which is (r)ead by default."""
  fp = open(path, mode)
  contents = fp.read()
  fp.close()
  return contents

# For replacing parts of contents in an existing file, with new content.
def file_substitute(path, contents, new_contents):
  """Replace the CONTENTS in the file at PATH using the NEW_CONTENTS"""
  fp = open(path, 'r')
  fcontent = fp.read()
  fp.close()
  fcontent = fcontent.replace(contents, new_contents)
  fp = open(path, 'w')
  fp.write(fcontent)
  fp.close()

# For creating blank new repositories
def create_repos(path):
  """Create a brand-new SVN repository at PATH.  If PATH does not yet
  exist, create it."""

  if not os.path.exists(path):
    os.makedirs(path) # this creates all the intermediate dirs, if neccessary

  opts = ("--bdb-txn-nosync",)
  if server_minor_version < 5:
    opts += ("--pre-1.5-compatible",)
  elif server_minor_version < 6:
    opts += ("--pre-1.6-compatible",)
  if fs_type is not None:
    opts += ("--fs-type=" + fs_type,)
  exit_code, stdout, stderr = run_command(svnadmin_binary, 1, 0, "create",
                                          path, *opts)

  # Skip tests if we can't create the repository.
  if stderr:
    for line in stderr:
      if line.find('Unknown FS type') != -1:
        raise Skip
    # If the FS type is known, assume the repos couldn't be created
    # (e.g. due to a missing 'svnadmin' binary).
    raise SVNRepositoryCreateFailure("".join(stderr).rstrip())

  # Allow unauthenticated users to write to the repos, for ra_svn testing.
  file_write(get_svnserve_conf_file_path(path),
             "[general]\nauth-access = write\n");
  if enable_sasl:
    file_append(get_svnserve_conf_file_path(path),
                "realm = svntest\n[sasl]\nuse-sasl = true\n")
  else:
    file_append(get_svnserve_conf_file_path(path), "password-db = <PASSWORD>")
    file_append(os.path.join(path, "conf", "passwd"),
                "[users]\njrandom = rayjandom\njconstant = rayjandom\n");

  if fs_type is None or fs_type == 'fsfs':
    # fsfs.conf file
    if config_file is not None:
      shutil.copy(config_file, get_fsfs_conf_file_path(path))

    # format file
    if fsfs_sharding is not None:
      def transform_line(line):
        if line.startswith('layout '):
          if fsfs_sharding > 0:
            line = 'layout sharded %d' % fsfs_sharding
          else:
            line = 'layout linear'
        return line

      # read it
      format_file_path = get_fsfs_format_file_path(path)
      contents = file_read(format_file_path, 'rb')

      # tweak it
      new_contents = "".join([transform_line(line) + "\n"
                              for line in contents.split("\n")])
      if new_contents[-1] == "\n":
        # we don't currently allow empty lines (\n\n) in the format file.
        new_contents = new_contents[:-1]

      # replace it
      os.chmod(format_file_path, 0666)
      file_write(format_file_path, new_contents, 'wb')

    # post-commit
    # Note that some tests (currently only commit_tests) create their own
    # post-commit hooks, which would override this one. :-(
    if fsfs_packing:
      # some tests chdir.
      abs_path = os.path.abspath(path)
      create_python_hook_script(get_post_commit_hook_path(abs_path),
          "import subprocess\n"
          "import sys\n"
          "command = %s\n"
          "sys.exit(subprocess.Popen(command).wait())\n"
          % repr([svnadmin_binary, 'pack', abs_path]))

  # make the repos world-writeable, for mod_dav_svn's sake.
  chmod_tree(path, 0666, 0666)

# For copying a repository
def copy_repos(src_path, dst_path, head_revision, ignore_uuid = 1):
  "Copy the repository SRC_PATH, with head revision HEAD_REVISION, to DST_PATH"

  # Do an svnadmin dump|svnadmin load cycle. Print a fake pipe command so that
  # the displayed CMDs can be run by hand
  os.environ['SVN_DBG_QUIET'] = 'y'
  create_repos(dst_path)
  dump_args = ['dump', src_path]
  load_args = ['load', dst_path]

  if ignore_uuid:
    load_args = load_args + ['--ignore-uuid']
  if verbose_mode:
    sys.stdout.write('CMD: %s%s | %s%s\n' % (os.path.basename(svnadmin_binary),
                                            ' '.join(dump_args),
                                            os.path.basename(svnadmin_binary),
                                            ' '.join(load_args)))
    sys.stdout.flush()
  start = time.time()

  dump_in, dump_out, dump_err, dump_kid = open_pipe(
    [svnadmin_binary] + dump_args)
  load_in, load_out, load_err, load_kid = open_pipe(
    [svnadmin_binary] + load_args,
    stdin=dump_out) # Attached to dump_kid

  stop = time.time()<commit_msg>Fix typo in command message that gets written to the test log. A space was missing between the command and its arguments: "svnadmindump" -> "svnadmin dump"<commit_after>#
#  main.py: a shared, automated test suite for Subversion
#
#  Subversion is a tool for revision control.
#  See http://subversion.tigris.org for more information.
#
# ====================================================================
#    Licensed to the Subversion Corporation (SVN Corp.) under one
#    or more contributor license agreements.  See the NOTICE file
#    distributed with this work for additional information
#    regarding copyright ownership.  The SVN Corp. 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.
######################################################################

import sys     # for argv[]
import os
import shutil  # for rmtree()
import re
import stat    # for ST_MODE
import subprocess
import copy    # for deepcopy()
import time    # for time()
import traceback # for print_exc()
import threading
try:
  # Python >=3.0
  import queue
  from urllib.parse import quote as urllib_parse_quote
  from urllib.parse import unquote as urllib_parse_unquote
except ImportError:
  # Python <3.0
  import Queue as queue
  from urllib import quote as urllib_parse_quote
  from urllib import unquote as urllib_parse_unquote

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

import svntest
from svntest import Failure
from svntest import Skip


######################################################################
#
#  HOW TO USE THIS MODULE:
#
#  Write a new python script that
#
#     1) imports this 'svntest' package
#
#     2) contains a number of related 'test' routines.  (Each test
#        routine should take no arguments, and return None on success
#        or throw a Failure exception on failure.  Each test should
#        also contain a short docstring.)
#
#     3) places all the tests into a list that begins with None.
#
#     4) calls svntest.main.client_test() on the list.
#
#  Also, your tests will probably want to use some of the common
#  routines in the 'Utilities' section below.
#
#####################################################################
# Global stuff

class SVNProcessTerminatedBySignal(Failure):
  "Exception raised if a spawned process segfaulted, aborted, etc."
  pass

class SVNLineUnequal(Failure):
  "Exception raised if two lines are unequal"
  pass

class SVNUnmatchedError(Failure):
  "Exception raised if an expected error is not found"
  pass

class SVNCommitFailure(Failure):
  "Exception raised if a commit failed"
  pass

class SVNRepositoryCopyFailure(Failure):
  "Exception raised if unable to copy a repository"
  pass

class SVNRepositoryCreateFailure(Failure):
  "Exception raised if unable to create a repository"
  pass

# Windows specifics
if sys.platform == 'win32':
  windows = True
  file_scheme_prefix = 'file:///'
  _exe = '.exe'
  _bat = '.bat'
else:
  windows = False
  file_scheme_prefix = 'file://'
  _exe = ''
  _bat = ''

# The location of our mock svneditor script.
if windows:
  svneditor_script = os.path.join(sys.path[0], 'svneditor.bat')
else:
  svneditor_script = os.path.join(sys.path[0], 'svneditor.py')

# Username and password used by the working copies
wc_author = 'jrandom'
wc_passwd = '<PASSWORD>'

# Username and password used by the working copies for "second user"
# scenarios
wc_author2 = 'jconstant' # use the same password as wc_author

# Set C locale for command line programs
os.environ['LC_ALL'] = 'C'

# This function mimics the Python 2.3 urllib function of the same name.
def pathname2url(path):
  """Convert the pathname PATH from the local syntax for a path to the form
  used in the path component of a URL. This does not produce a complete URL.
  The return value will already be quoted using the quote() function."""
  return urllib_parse_quote(path.replace('\\', '/'))

# This function mimics the Python 2.3 urllib function of the same name.
def url2pathname(path):
  """Convert the path component PATH from an encoded URL to the local syntax
  for a path. This does not accept a complete URL. This function uses
  unquote() to decode PATH."""
  return os.path.normpath(urllib_parse_unquote(path))

######################################################################
# Global variables set during option parsing.  These should not be used
# until the variable command_line_parsed has been set to True, as is
# done in run_tests below.
command_line_parsed = False

# The locations of the svn, svnadmin and svnlook binaries, relative to
# the only scripts that import this file right now (they live in ../).
# Use --bin to override these defaults.
svn_binary = os.path.abspath('../../svn/svn' + _exe)
svnadmin_binary = os.path.abspath('../../svnadmin/svnadmin' + _exe)
svnlook_binary = os.path.abspath('../../svnlook/svnlook' + _exe)
svnsync_binary = os.path.abspath('../../svnsync/svnsync' + _exe)
svnversion_binary = os.path.abspath('../../svnversion/svnversion' + _exe)
svndumpfilter_binary = os.path.abspath('../../svndumpfilter/svndumpfilter' + \
                                       _exe)
entriesdump_binary = os.path.abspath('entries-dump' + _exe)

# Global variable indicating if we want verbose output, that is,
# details of what commands each test does as it does them.  This is
# incompatible with quiet_mode.
verbose_mode = False

# Global variable indicating if we want quiet output, that is, don't
# show PASS, XFAIL, or SKIP notices, but do show FAIL and XPASS.  This
# is incompatible with verbose_mode.
quiet_mode = False

# Global variable indicating if we want test data cleaned up after success
cleanup_mode = False

# Global variable indicating if svnserve should use Cyrus SASL
enable_sasl = False

# Global variable indicating that SVNKit binaries should be used
use_jsvn = False

# Global variable indicating which DAV library, if any, is in use
# ('neon', 'serf')
http_library = None

# Global variable: Number of shards to use in FSFS
# 'None' means "use FSFS's default"
fsfs_sharding = None

# Global variable: automatically pack FSFS repositories after every commit
fsfs_packing = None

# Configuration file (copied into FSFS fsfs.conf).
config_file = None

# Global variable indicating what the minor version of the server
# tested against is (4 for 1.4.x, for example).
server_minor_version = 7

# Global variable indicating if this is a child process and no cleanup
# of global directories is needed.
is_child_process = False

# Global URL to testing area.  Default to ra_local, current working dir.
test_area_url = file_scheme_prefix + pathname2url(os.path.abspath(os.getcwd()))

# Location to the pristine repository, will be calculated from test_area_url
# when we know what the user specified for --url.
pristine_url = None

# Global variable indicating the FS type for repository creations.
fs_type = None

# End of command-line-set global variables.
######################################################################

# All temporary repositories and working copies are created underneath
# this dir, so there's one point at which to mount, e.g., a ramdisk.
work_dir = "svn-test-work"

# Constant for the merge info property.
SVN_PROP_MERGEINFO = "svn:mergeinfo"

# Where we want all the repositories and working copies to live.
# Each test will have its own!
general_repo_dir = os.path.join(work_dir, "repositories")
general_wc_dir = os.path.join(work_dir, "working_copies")

# temp directory in which we will create our 'pristine' local
# repository and other scratch data.  This should be removed when we
# quit and when we startup.
temp_dir = os.path.join(work_dir, 'local_tmp')

# (derivatives of the tmp dir.)
pristine_dir = os.path.join(temp_dir, "repos")
greek_dump_dir = os.path.join(temp_dir, "greekfiles")
default_config_dir = os.path.abspath(os.path.join(temp_dir, "config"))

#
# Our pristine greek-tree state.
#
# If a test wishes to create an "expected" working-copy tree, it should
# call main.greek_state.copy().  That method will return a copy of this
# State object which can then be edited.
#
_item = svntest.wc.StateItem
greek_state = svntest.wc.State('', {
  'iota'        : _item("This is the file 'iota'.\n"),
  'A'           : _item(),
  'A/mu'        : _item("This is the file 'mu'.\n"),
  'A/B'         : _item(),
  'A/B/lambda'  : _item("This is the file 'lambda'.\n"),
  'A/B/E'       : _item(),
  'A/B/E/alpha' : _item("This is the file 'alpha'.\n"),
  'A/B/E/beta'  : _item("This is the file 'beta'.\n"),
  'A/B/F'       : _item(),
  'A/C'         : _item(),
  'A/D'         : _item(),
  'A/D/gamma'   : _item("This is the file 'gamma'.\n"),
  'A/D/G'       : _item(),
  'A/D/G/pi'    : _item("This is the file 'pi'.\n"),
  'A/D/G/rho'   : _item("This is the file 'rho'.\n"),
  'A/D/G/tau'   : _item("This is the file 'tau'.\n"),
  'A/D/H'       : _item(),
  'A/D/H/chi'   : _item("This is the file 'chi'.\n"),
  'A/D/H/psi'   : _item("This is the file 'psi'.\n"),
  'A/D/H/omega' : _item("This is the file 'omega'.\n"),
  })


######################################################################
# Utilities shared by the tests
def wrap_ex(func):
  "Wrap a function, catch, print and ignore exceptions"
  def w(*args, **kwds):
    try:
      return func(*args, **kwds)
    except Failure, ex:
      if ex.__class__ != Failure or ex.args:
        ex_args = str(ex)
        if ex_args:
          print('EXCEPTION: %s: %s' % (ex.__class__.__name__, ex_args))
        else:
          print('EXCEPTION: %s' % ex.__class__.__name__)
  return w

def setup_development_mode():
  "Wraps functions in module actions"
  l = [ 'run_and_verify_svn',
        'run_and_verify_svnversion',
        'run_and_verify_load',
        'run_and_verify_dump',
        'run_and_verify_checkout',
        'run_and_verify_export',
        'run_and_verify_update',
        'run_and_verify_merge',
        'run_and_verify_merge2',
        'run_and_verify_switch',
        'run_and_verify_commit',
        'run_and_verify_unquiet_status',
        'run_and_verify_status',
        'run_and_verify_diff_summarize',
        'run_and_verify_diff_summarize_xml',
        'run_and_validate_lock']

  for func in l:
    setattr(svntest.actions, func, wrap_ex(getattr(svntest.actions, func)))

def get_admin_name():
  "Return name of SVN administrative subdirectory."

  if (windows or sys.platform == 'cygwin') \
      and 'SVN_ASP_DOT_NET_HACK' in os.environ:
    return '_svn'
  else:
    return '.svn'

def get_start_commit_hook_path(repo_dir):
  "Return the path of the start-commit-hook conf file in REPO_DIR."

  return os.path.join(repo_dir, "hooks", "start-commit")


def get_pre_commit_hook_path(repo_dir):
  "Return the path of the pre-commit-hook conf file in REPO_DIR."

  return os.path.join(repo_dir, "hooks", "pre-commit")


def get_post_commit_hook_path(repo_dir):
  "Return the path of the post-commit-hook conf file in REPO_DIR."

  return os.path.join(repo_dir, "hooks", "post-commit")

def get_pre_revprop_change_hook_path(repo_dir):
  "Return the path of the pre-revprop-change hook script in REPO_DIR."

  return os.path.join(repo_dir, "hooks", "pre-revprop-change")

def get_svnserve_conf_file_path(repo_dir):
  "Return the path of the svnserve.conf file in REPO_DIR."

  return os.path.join(repo_dir, "conf", "svnserve.conf")

def get_fsfs_conf_file_path(repo_dir):
  "Return the path of the fsfs.conf file in REPO_DIR."

  return os.path.join(repo_dir, "db", "fsfs.conf")

def get_fsfs_format_file_path(repo_dir):
  "Return the path of the format file in REPO_DIR."

  return os.path.join(repo_dir, "db", "format")

# Run any binary, logging the command line and return code
def run_command(command, error_expected, binary_mode=0, *varargs):
  """Run COMMAND with VARARGS; return exit code as int; stdout, stderr
  as lists of lines.
  If ERROR_EXPECTED is None, any stderr also will be printed."""

  return run_command_stdin(command, error_expected, binary_mode,
                           None, *varargs)

# A regular expression that matches arguments that are trivially safe
# to pass on a command line without quoting on any supported operating
# system:
_safe_arg_re = re.compile(r'^[A-Za-z\d\.\_\/\-\:\@]+$')

def _quote_arg(arg):
  """Quote ARG for a command line.

  Simply surround every argument in double-quotes unless it contains
  only universally harmless characters.

  WARNING: This function cannot handle arbitrary command-line
  arguments.  It can easily be confused by shell metacharacters.  A
  perfect job would be difficult and OS-dependent (see, for example,
  http://msdn.microsoft.com/library/en-us/vccelng/htm/progs_12.asp).
  In other words, this function is just good enough for what we need
  here."""

  arg = str(arg)
  if _safe_arg_re.match(arg):
    return arg
  else:
    if os.name != 'nt':
      arg = arg.replace('$', '\$')
    return '"%s"' % (arg,)

def open_pipe(command, stdin=None, stdout=None, stderr=None):
  """Opens a subprocess.Popen pipe to COMMAND using STDIN,
  STDOUT, and STDERR.

  Returns (infile, outfile, errfile, waiter); waiter
  should be passed to wait_on_pipe."""
  command = [str(x) for x in command]

  # On Windows subprocess.Popen() won't accept a Python script as
  # a valid program to execute, rather it wants the Python executable.
  if (sys.platform == 'win32') and (command[0].endswith('.py')):
    command.insert(0, sys.executable)

  # Quote only the arguments on Windows.  Later versions of subprocess,
  # 2.5.2+ confirmed, don't require this quoting, but versions < 2.4.3 do.
  if (sys.platform == 'win32'):
    args = command[1:]
    args = ' '.join([_quote_arg(x) for x in args])
    command = command[0] + ' ' + args

  if not stdin:
    stdin = subprocess.PIPE
  if not stdout:
    stdout = subprocess.PIPE
  if not stderr:
    stderr = subprocess.PIPE

  p = subprocess.Popen(command,
                       stdin=stdin,
                       stdout=stdout,
                       stderr=stderr,
                       close_fds=not windows)
  return p.stdin, p.stdout, p.stderr, (p, command)

def wait_on_pipe(waiter, binary_mode, stdin=None):
  """Waits for KID (opened with open_pipe) to finish, dying
  if it does.  If kid fails create an error message containing
  any stdout and stderr from the kid.  Returns kid's exit code,
  stdout and stderr (the latter two as lists)."""
  if waiter is None:
    return

  kid, command = waiter
  stdout, stderr = kid.communicate(stdin)
  exit_code = kid.returncode

  # Normalize Windows line endings if in text mode.
  if windows and not binary_mode:
    stdout = stdout.replace('\r\n', '\n')
    stderr = stderr.replace('\r\n', '\n')

  # Convert output strings to lists.
  stdout_lines = stdout.splitlines(True)
  stderr_lines = stderr.splitlines(True)

  if exit_code < 0:
    exit_signal = os.WTERMSIG(-exit_code)
    if stdout_lines is not None:
      sys.stdout.write("".join(stdout_lines))
    if stderr_lines is not None:
      sys.stderr.write("".join(stderr_lines))
    if verbose_mode:
      # show the whole path to make it easier to start a debugger
      sys.stderr.write("CMD: %s terminated by signal %d\n"
                       % (' '.join(command), exit_signal))
    raise SVNProcessTerminatedBySignal
  else:
    if exit_code and verbose_mode:
      sys.stderr.write("CMD: %s exited with %d\n"
                       % (' '.join(command), exit_code))
    return stdout_lines, stderr_lines, exit_code

# Run any binary, supplying input text, logging the command line
def spawn_process(command, binary_mode=0, stdin_lines=None, *varargs):
  if stdin_lines and not isinstance(stdin_lines, list):
    raise TypeError("stdin_lines should have list type")

  # Log the command line
  if verbose_mode and not command.endswith('.py'):
    sys.stdout.write('CMD: %s %s\n' % (os.path.basename(command),
                                      ' '.join([_quote_arg(x) for x in varargs])))
    sys.stdout.flush()

  infile, outfile, errfile, kid = open_pipe([command] + list(varargs))

  if stdin_lines:
    for x in stdin_lines:
      infile.write(x)

  stdout_lines, stderr_lines, exit_code = wait_on_pipe(kid, binary_mode)
  infile.close()

  outfile.close()
  errfile.close()

  return exit_code, stdout_lines, stderr_lines

def run_command_stdin(command, error_expected, binary_mode=0,
                      stdin_lines=None, *varargs):
  """Run COMMAND with VARARGS; input STDIN_LINES (a list of strings
  which should include newline characters) to program via stdin - this
  should not be very large, as if the program outputs more than the OS
  is willing to buffer, this will deadlock, with both Python and
  COMMAND waiting to write to each other for ever.
  Return exit code as int; stdout, stderr as lists of lines.
  If ERROR_EXPECTED is None, any stderr also will be printed."""

  if verbose_mode:
    start = time.time()

  exit_code, stdout_lines, stderr_lines = spawn_process(command,
                                                        binary_mode,
                                                        stdin_lines,
                                                        *varargs)

  if verbose_mode:
    stop = time.time()
    print('<TIME = %.6f>' % (stop - start))
    for x in stdout_lines:
      sys.stdout.write(x)
    for x in stderr_lines:
      sys.stdout.write(x)

  if (not error_expected) and (stderr_lines):
    if not verbose_mode:
      for x in stderr_lines:
        sys.stdout.write(x)
    raise Failure

  return exit_code, \
         [line for line in stdout_lines if not line.startswith("DBG:")], \
         stderr_lines

def create_config_dir(cfgdir, config_contents=None, server_contents=None):
  "Create config directories and files"

  # config file names
  cfgfile_cfg = os.path.join(cfgdir, 'config')
  cfgfile_srv = os.path.join(cfgdir, 'servers')

  # create the directory
  if not os.path.isdir(cfgdir):
    os.makedirs(cfgdir)

  # define default config file contents if none provided
  if config_contents is None:
    config_contents = """
#
[auth]
password-stores =

[miscellany]
interactive-conflicts = false
"""

  # define default server file contents if none provided
  if server_contents is None:
    http_library_str = ""
    if http_library:
      http_library_str = "http-library=%s" % (http_library)
    server_contents = """
#
[global]
%s
store-plaintext-passwords=yes
store-passwords=yes
""" % (http_library_str)

  file_write(cfgfile_cfg, config_contents)
  file_write(cfgfile_srv, server_contents)

def _with_config_dir(args):
  if '--config-dir' in args:
    return args
  else:
    return args + ('--config-dir', default_config_dir)

def _with_auth(args):
  assert '--password' not in args
  args = args + ('--password', <PASSWORD>,
                 '--no-auth-cache' )
  if '--username' in args:
    return args
  else:
    return args + ('--username', wc_author )

# For running subversion and returning the output
def run_svn(error_expected, *varargs):
  """Run svn with VARARGS; return exit code as int; stdout, stderr as
  lists of lines.
  If ERROR_EXPECTED is None, any stderr also will be printed.  If
  you're just checking that something does/doesn't come out of
  stdout/stderr, you might want to use actions.run_and_verify_svn()."""
  return run_command(svn_binary, error_expected, 0,
                     *(_with_auth(_with_config_dir(varargs))))

# For running svnadmin.  Ignores the output.
def run_svnadmin(*varargs):
  """Run svnadmin with VARARGS, returns exit code as int; stdout, stderr as
  list of lines."""
  return run_command(svnadmin_binary, 1, 0, *varargs)

# For running svnlook.  Ignores the output.
def run_svnlook(*varargs):
  """Run svnlook with VARARGS, returns exit code as int; stdout, stderr as
  list of lines."""
  return run_command(svnlook_binary, 1, 0, *varargs)

def run_svnsync(*varargs):
  """Run svnsync with VARARGS, returns exit code as int; stdout, stderr as
  list of lines."""
  return run_command(svnsync_binary, 1, 0, *(_with_config_dir(varargs)))

def run_svnversion(*varargs):
  """Run svnversion with VARARGS, returns exit code as int; stdout, stderr
  as list of lines."""
  return run_command(svnversion_binary, 1, 0, *varargs)

def run_entriesdump(path):
  """Run the entries-dump helper, returning a dict of Entry objects."""
  # use spawn_process rather than run_command to avoid copying all the data
  # to stdout in verbose mode.
  exit_code, stdout_lines, stderr_lines = spawn_process(entriesdump_binary,
                                                        0, None, path)
  if verbose_mode:
    ### finish the CMD output
    print
  if exit_code or stderr_lines:
    ### report on this? or continue to just skip it?
    return None

  class Entry(object):
    pass
  entries = { }
  exec(''.join([line for line in stdout_lines if not line.startswith("DBG:")]))
  return entries


# Chmod recursively on a whole subtree
def chmod_tree(path, mode, mask):
  for dirpath, dirs, files in os.walk(path):
    for name in dirs + files:
      fullname = os.path.join(dirpath, name)
      if not os.path.islink(fullname):
        new_mode = (os.stat(fullname)[stat.ST_MODE] & ~mask) | mode
        os.chmod(fullname, new_mode)

# For clearing away working copies
def safe_rmtree(dirname, retry=0):
  "Remove the tree at DIRNAME, making it writable first"
  def rmtree(dirname):
    chmod_tree(dirname, 0666, 0666)
    shutil.rmtree(dirname)

  if not os.path.exists(dirname):
    return

  if retry:
    for delay in (0.5, 1, 2, 4):
      try:
        rmtree(dirname)
        break
      except:
        time.sleep(delay)
    else:
      rmtree(dirname)
  else:
    rmtree(dirname)

# For making local mods to files
def file_append(path, new_text):
  "Append NEW_TEXT to file at PATH"
  open(path, 'a').write(new_text)

# Append in binary mode
def file_append_binary(path, new_text):
  "Append NEW_TEXT to file at PATH in binary mode"
  open(path, 'ab').write(new_text)

# For creating new files, and making local mods to existing files.
def file_write(path, contents, mode='w'):
  """Write the CONTENTS to the file at PATH, opening file using MODE,
  which is (w)rite by default."""
  open(path, mode).write(contents)

# For reading the contents of a file
def file_read(path, mode = 'r'):
  """Return the contents of the file at PATH, opening file using MODE,
  which is (r)ead by default."""
  fp = open(path, mode)
  contents = fp.read()
  fp.close()
  return contents

# For replacing parts of contents in an existing file, with new content.
def file_substitute(path, contents, new_contents):
  """Replace the CONTENTS in the file at PATH using the NEW_CONTENTS"""
  fp = open(path, 'r')
  fcontent = fp.read()
  fp.close()
  fcontent = fcontent.replace(contents, new_contents)
  fp = open(path, 'w')
  fp.write(fcontent)
  fp.close()

# For creating blank new repositories
def create_repos(path):
  """Create a brand-new SVN repository at PATH.  If PATH does not yet
  exist, create it."""

  if not os.path.exists(path):
    os.makedirs(path) # this creates all the intermediate dirs, if neccessary

  opts = ("--bdb-txn-nosync",)
  if server_minor_version < 5:
    opts += ("--pre-1.5-compatible",)
  elif server_minor_version < 6:
    opts += ("--pre-1.6-compatible",)
  if fs_type is not None:
    opts += ("--fs-type=" + fs_type,)
  exit_code, stdout, stderr = run_command(svnadmin_binary, 1, 0, "create",
                                          path, *opts)

  # Skip tests if we can't create the repository.
  if stderr:
    for line in stderr:
      if line.find('Unknown FS type') != -1:
        raise Skip
    # If the FS type is known, assume the repos couldn't be created
    # (e.g. due to a missing 'svnadmin' binary).
    raise SVNRepositoryCreateFailure("".join(stderr).rstrip())

  # Allow unauthenticated users to write to the repos, for ra_svn testing.
  file_write(get_svnserve_conf_file_path(path),
             "[general]\nauth-access = write\n");
  if enable_sasl:
    file_append(get_svnserve_conf_file_path(path),
                "realm = svntest\n[sasl]\nuse-sasl = true\n")
  else:
    file_append(get_svnserve_conf_file_path(path), "password-db = passwd\n")
    file_append(os.path.join(path, "conf", "passwd"),
                "[users]\njrandom = rayjandom\njconstant = rayjandom\n");

  if fs_type is None or fs_type == 'fsfs':
    # fsfs.conf file
    if config_file is not None:
      shutil.copy(config_file, get_fsfs_conf_file_path(path))

    # format file
    if fsfs_sharding is not None:
      def transform_line(line):
        if line.startswith('layout '):
          if fsfs_sharding > 0:
            line = 'layout sharded %d' % fsfs_sharding
          else:
            line = 'layout linear'
        return line

      # read it
      format_file_path = get_fsfs_format_file_path(path)
      contents = file_read(format_file_path, 'rb')

      # tweak it
      new_contents = "".join([transform_line(line) + "\n"
                              for line in contents.split("\n")])
      if new_contents[-1] == "\n":
        # we don't currently allow empty lines (\n\n) in the format file.
        new_contents = new_contents[:-1]

      # replace it
      os.chmod(format_file_path, 0666)
      file_write(format_file_path, new_contents, 'wb')

    # post-commit
    # Note that some tests (currently only commit_tests) create their own
    # post-commit hooks, which would override this one. :-(
    if fsfs_packing:
      # some tests chdir.
      abs_path = os.path.abspath(path)
      create_python_hook_script(get_post_commit_hook_path(abs_path),
          "import subprocess\n"
          "import sys\n"
          "command = %s\n"
          "sys.exit(subprocess.Popen(command).wait())\n"
          % repr([svnadmin_binary, 'pack', abs_path]))

  # make the repos world-writeable, for mod_dav_svn's sake.
  chmod_tree(path, 0666, 0666)

# For copying a repository
def copy_repos(src_path, dst_path, head_revision, ignore_uuid = 1):
  "Copy the repository SRC_PATH, with head revision HEAD_REVISION, to DST_PATH"

  # Do an svnadmin dump|svnadmin load cycle. Print a fake pipe command so that
  # the displayed CMDs can be run by hand
  os.environ['SVN_DBG_QUIET'] = 'y'
  create_repos(dst_path)
  dump_args = ['dump', src_path]
  load_args = ['load', dst_path]

  if ignore_uuid:
    load_args = load_args + ['--ignore-uuid']
  if verbose_mode:
    sys.stdout.write('CMD: %s %s | %s %s\n' %
                     (os.path.basename(svnadmin_binary), ' '.join(dump_args),
                      os.path.basename(svnadmin_binary), ' '.join(load_args)))
    sys.stdout.flush()
  start = time.time()

  dump_in, dump_out, dump_err, dump_kid = open_pipe(
    [svnadmin_binary] + dump_args)
  load_in, load_out, load_err, load_kid = open_pipe(
    [svnadmin_binary] + load_args,
    stdin=dump_out) # Attached to dump_kid

  stop = time.time()
{'fraction_non_alphanumeric': 0.11953783813382982, 'fraction_numerical': 0.0036140619866389225, 'mean_word_length': 3.3309881422924903, 'pattern_counts': {'":': 2, '<': 19, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 32, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6802758', 'n_tokens_mistral': 17887, 'n_tokens_neox': 17252, 'n_words': 5675}