actions.py

subversion-1.4.3-1.tar.gz 配置svn的源码

#!/usr/bin/env python
#
#  actions.py:  routines that actually run the svn client.
#
#  Subversion is a tool for revision control.
#  See http://subversion.tigris.org for more information.
#
# ====================================================================
# Copyright (c) 2000-2006 CollabNet.  All rights reserved.
#
# This software is licensed as described in the file COPYING, which
# you should have received as part of this distribution.  The terms
# are also available at http://subversion.tigris.org/license-1.html.
# If newer versions of this license are posted there, you may use a
# newer version instead, at your option.
#
######################################################################

import os.path, shutil, string, re, sys, errno

import main, tree, wc  # general svntest routines in this module.
from svntest import Failure, SVNAnyOutput

class SVNUnexpectedOutput(Failure):
  """Exception raised if an invocation of svn results in unexpected
  output of any kind."""
  pass

class SVNUnexpectedStdout(SVNUnexpectedOutput):
  """Exception raised if an invocation of svn results in unexpected
  output on STDOUT."""
  pass

class SVNUnexpectedStderr(SVNUnexpectedOutput):
  """Exception raised if an invocation of svn results in unexpected
  output on STDERR."""
  pass

class SVNExpectedStdout(SVNUnexpectedOutput):
  """Exception raised if an invocation of svn results in no output on
  STDOUT when output was expected."""
  pass

class SVNExpectedStderr(SVNUnexpectedOutput):
  """Exception raised if an invocation of svn results in no output on
  STDERR when output was expected."""
  pass

class SVNIncorrectDatatype(SVNUnexpectedOutput):
  """Exception raised if invalid input is passed to the
  run_and_verify_* API"""
  pass


######################################################################
# Used by every test, so that they can run independently of
# one another.  The first time it's run, it executes 'svnadmin' to
# create a repository and then 'svn imports' the greek tree.
# Thereafter, every time this routine is called, it recursively copies
# the `pristine repos' to a new location.

def guarantee_greek_repository(path):
  """Guarantee that a local svn repository exists at PATH, containing
  nothing but the greek-tree at revision 1."""

  if path == main.pristine_dir:
    print "ERROR:  attempt to overwrite the pristine repos!  Aborting."
    sys.exit(1)

  # If there's no pristine repos, create one.
  if not os.path.exists(main.pristine_dir):
    main.create_repos(main.pristine_dir)

    # dump the greek tree to disk.
    main.greek_state.write_to_disk(main.greek_dump_dir)

    # build a URL for doing an import.
    url = main.test_area_url + '/' + main.pristine_dir
    if main.windows == 1:
      url = string.replace(url, '\\', '/')

    # import the greek tree, using l:foo/p:bar
    ### todo: svn should not be prompting for auth info when using
    ### repositories with no auth/auth requirements
    output, errput = main.run_svn(None, 'import',
                                  '--username', main.wc_author,
                                  '--password', main.wc_passwd,
                                  '-m', 'Log message for revision 1.',
                                  main.greek_dump_dir, url)

    # check for any errors from the import
    if len(errput):
      display_lines("Errors during initial 'svn import':",
                    'STDERR', None, errput)
      sys.exit(1)

    # verify the printed output of 'svn import'.
    lastline = string.strip(output.pop())
    cm = re.compile ("(Committed|Imported) revision [0-9]+.")
    match = cm.search (lastline)
    if not match:
      print "ERROR:  import did not succeed, while creating greek repos."
      print "The final line from 'svn import' was:"
      print lastline
      sys.exit(1)
    output_tree = tree.build_tree_from_commit(output)

    ### due to path normalization in the .old_tree() method, we cannot
    ### prepend the necessary '.' directory. thus, let's construct an old
    ### tree manually from the greek_state.
    output_list = []
    for greek_path in main.greek_state.desc.keys():
      output_list.append([ os.path.join(main.greek_dump_dir, greek_path),
                           None, {}, {'verb' : 'Adding'}])
    expected_output_tree = tree.build_generic_tree(output_list)

    try:
      tree.compare_trees(output_tree, expected_output_tree)
    except tree.SVNTreeUnequal:
      display_trees("ERROR:  output of import command is unexpected.",
                    'OUTPUT TREE', expected_output_tree, output_tree)
      sys.exit(1)

  # Now that the pristine repos exists, copy it to PATH.
  main.safe_rmtree(path)
  if main.copy_repos(main.pristine_dir, path, 1):
    print "ERROR:  copying repository failed."
    sys.exit(1)

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

def run_and_verify_svnversion(message, wc_dir, repo_url,
                              expected_stdout, expected_stderr):
  "Run svnversion command and check its output"

  out, err = main.run_svnversion(wc_dir, repo_url)

  if type(expected_stdout) is type([]):
    compare_and_display_lines(message, 'STDOUT', expected_stdout, out)
  elif expected_stdout == SVNAnyOutput:
    if len(out) == 0:
      if message is not None: print message
      raise SVNExpectedStdout
  elif expected_stdout is not None:
    raise SVNIncorrectDatatype("Unexpected specification for stdout data")

  if type(expected_stderr) is type([]):
    compare_and_display_lines(message, 'STDERR', expected_stderr, err)
  elif expected_stderr == SVNAnyOutput:
    if len(err) == 0:
      if message is not None: print message
      raise SVNExpectedStderr
  else:
    raise SVNIncorrectDatatype("Unexpected specification for stderr data")
  return out, err


def run_and_verify_svn(message, expected_stdout, expected_stderr, *varargs):
  """Invokes main.run_svn with *VARARGS, return stdout and stderr as
  lists of lines.  For both EXPECTED_STDOUT and EXPECTED_STDERR, do this:

     - If it is an array of strings, invoke compare_and_display_lines()
       on MESSAGE, the expected output, and the actual output.

     - If it is a single string, invoke match_or_fail() on MESSAGE,
       the expected output, and the actual output.

  If EXPECTED_STDOUT is None, do not check stdout.
  EXPECTED_STDERR may not be None.

  If a comparison function fails, it will raise an error."""
  ### TODO catch and throw particular exceptions from above

  if expected_stderr is None:
    raise SVNIncorrectDatatype("expected_stderr must not be None")

  want_err = None
  if expected_stderr is not None and expected_stderr is not []:
    want_err = 1

  out, err = main.run_svn(want_err, *varargs)

  for (expected, actual, output_type, raisable) in (
      (expected_stderr, err, 'stderr', SVNExpectedStderr),
      (expected_stdout, out, 'stdout', SVNExpectedStdout)):
    if type(expected) is type([]):
      compare_and_display_lines(message, output_type.upper(), expected, actual)
    elif type(expected) is type(''):
      match_or_fail(message, output_type.upper(), expected, actual)
    elif expected == SVNAnyOutput:
      if len(actual) == 0:
        if message is not None: print message
        raise raisable
    elif expected is not None:
      raise SVNIncorrectDatatype("Unexpected type for %s data" % output_type)

  return out, err


def run_and_verify_load(repo_dir, dump_file_content):
  "Runs 'svnadmin load' and reports any errors."
  expected_stderr = []
  output, errput = \
          main.run_command_stdin(
    "%s load --force-uuid --quiet %s" % (main.svnadmin_binary, repo_dir),
    expected_stderr, 1, dump_file_content)
  if expected_stderr:
    actions.compare_and_display_lines(
      "Standard error output", "STDERR", expected_stderr, errput)


def run_and_verify_dump(repo_dir):
  "Runs 'svnadmin dump' and reports any errors, returning the dump content."
  output, errput = main.run_svnadmin('dump', repo_dir)
  if not output:
    raise svntest.actions.SVNUnexpectedStdout("Missing stdout")
  if not errput:
    raise svntest.actions.SVNUnexpectedStderr("Missing stderr")

  return output

######################################################################
# Subversion Actions
#
# These are all routines that invoke 'svn' in particular ways, and
# then verify the results by comparing expected trees with actual
# trees.
#
# For all the functions below, the OUTPUT_TREE and DISK_TREE args need
# to be created by feeding carefully constructed lists to
# tree.build_generic_tree().  A STATUS_TREE can be built by
# hand, or by editing the tree returned by get_virginal_state().


def run_and_verify_checkout(URL, wc_dir_name, output_tree, disk_tree,
                            singleton_handler_a = None,
                            a_baton = None,
                            singleton_handler_b = None,
                            b_baton = None):
  """Checkout the URL into a new directory WC_DIR_NAME.

  The subcommand output will be verified against OUTPUT_TREE,
  and the working copy itself will be verified against DISK_TREE.
  SINGLETON_HANDLER_A and SINGLETON_HANDLER_B will be passed to
  tree.compare_trees - see that function's doc string for more details.
  Returns if successful and raise on failure."""

  if isinstance(output_tree, wc.State):
    output_tree = output_tree.old_tree()
  if isinstance(disk_tree, wc.State):
    disk_tree = disk_tree.old_tree()

  # Remove dir if it's already there.
  main.safe_rmtree(wc_dir_name)

  # Checkout and make a tree of the output, using l:foo/p:bar
  ### todo: svn should not be prompting for auth info when using
  ### repositories with no auth/auth requirements
  output, errput = main.run_svn (None, 'co',
                                 '--username', main.wc_author,
                                 '--password', main.wc_passwd,
                                 URL, wc_dir_name)
  mytree = tree.build_tree_from_checkout (output)

  # Verify actual output against expected output.
  tree.compare_trees (mytree, output_tree)

  # Create a tree by scanning the working copy
  mytree = tree.build_tree_from_wc (wc_dir_name)

  # Verify expected disk against actual disk.
  tree.compare_trees (mytree, disk_tree,
                      singleton_handler_a, a_baton,
                      singleton_handler_b, b_baton)


def run_and_verify_export(URL, export_dir_name, output_tree, disk_tree,
                          singleton_handler_a = None,
                          a_baton = None,
                          singleton_handler_b = None,
                          b_baton = None,
                          *args):
  """Export the URL into a new directory WC_DIR_NAME.

  The subcommand output will be verified against OUTPUT_TREE,
  and the exported copy itself will be verified against DISK_TREE.
  SINGLETON_HANDLER_A and SINGLETON_HANDLER_B will be passed to
  tree.compare_trees - see that function's doc string for more details.
  Returns if successful and raise on failure."""

  if isinstance(output_tree, wc.State):
    output_tree = output_tree.old_tree()
  if isinstance(disk_tree, wc.State):
    disk_tree = disk_tree.old_tree()

  # Export and make a tree of the output, using l:foo/p:bar
  ### todo: svn should not be prompting for auth info when using
  ### repositories with no auth/auth requirements
  output, errput = main.run_svn (None, 'export',
                                 '--username', main.wc_author,
                                 '--password', main.wc_passwd,
                                 URL, export_dir_name, *args)
  mytree = tree.build_tree_from_checkout (output)

  # Verify actual output against expected output.
  tree.compare_trees (mytree, output_tree)

  # Create a tree by scanning the working copy.  Don't ignore
  # the .svn directories so that we generate an error if they
  # happen to show up.