svn2svn/shell.py

   1 """ Shell functions """
   2
   3 from svn2svn import ui
   4 from errors import ExternalCommandFailed
   5
   6 import os
   7 import locale
   8 import time
   9 import shutil
  10 import stat
  11 import sys
  12 import traceback
  13 import re
  14 from datetime import datetime
  15 from subprocess import Popen, PIPE, STDOUT
  16
  17 try:
  18     import commands
  19 except ImportError:
  20     commands = None
  21
  22
  23 # Windows compatibility code by Bill Baxter
  24 if os.name == "nt":
  25     def find_program(name):
  26         """
  27         Find the name of the program for Popen.
  28         Windows is finnicky about having the complete file name. Popen
  29         won't search the %PATH% for you automatically.
  30         (Adapted from ctypes.find_library)
  31         """
  32         # See MSDN for the REAL search order.
  33         base, ext = os.path.splitext(name)
  34         if ext:
  35             exts = [ext]
  36         else:
  37             exts = ['.bat', '.exe']
  38         for directory in os.environ['PATH'].split(os.pathsep):
  39             for e in exts:
  40                 fname = os.path.join(directory, base + e)
  41                 if os.path.exists(fname):
  42                     return fname
  43         return name
  44 else:
  45     def find_program(name):
  46         """
  47         Find the name of the program for Popen.
  48         On Unix, popen isn't picky about having absolute paths.
  49         """
  50         return name
  51
  52
  53 def _rmtree_error_handler(func, path, exc_info):
  54     """
  55     Error handler for rmtree. Helps removing the read-only protection under
  56     Windows (and others?).
  57     Adapted from http://www.proaxis.com/~darkwing/hot-backup.py
  58     and http://patchwork.ozlabs.org/bazaar-ng/patch?id=4243
  59     """
  60     if func in (os.remove, os.rmdir) and os.path.exists(path):
  61         # Change from read-only to writeable
  62         os.chmod(path, os.stat(path).st_mode | stat.S_IWRITE)
  63         func(path)
  64     else:
  65         # something else must be wrong...
  66         raise
  67
  68 def rmtree(path):
  69     """
  70     Wrapper around shutil.rmtree(), to provide more error-resistent behaviour.
  71     """
  72     return shutil.rmtree(path, False, _rmtree_error_handler)
  73
  74
  75 # Make sure we do not get localized output from the Subversion
  76 # command line client.
  77 os.environ['LC_MESSAGES'] = 'C'
  78
  79 locale_encoding = locale.getpreferredencoding()
  80
  81 def get_encoding():
  82     return locale_encoding
  83
  84 def shell_quote(s):
  85     # No need to wrap "safe" strings in quotes
  86     if re.compile('^[A-Za-z0-9=-]+$').match(s):
  87         return s
  88     if os.name == "nt":
  89         q = '"'
  90     else:
  91         q = "'"
  92     return q + s.replace('\\', '\\\\').replace("'", "'\"'\"'") + q
  93
  94 def _run_raw_command(cmd, args, fail_if_stderr=False, no_fail=False):
  95     cmd_string = "%s %s" % (cmd,  " ".join(map(shell_quote, args)))
  96     color = 'BLUE_B'
  97     if cmd == 'svn' and args[0] in ['status', 'st', 'log', 'info', 'list', 'proplist', 'propget', 'update', 'up', 'cleanup', 'revert']:
  98         # Show status-only commands (commands which make no changes to WC) in dim-blue
  99         color = 'BLUE'
 100     ui.status("$ %s", cmd_string, level=ui.EXTRA, color=color)
 101     try:
 102         pipe = Popen([cmd] + args, executable=cmd, stdout=PIPE, stderr=PIPE)
 103     except OSError:
 104         etype, value = sys.exc_info()[:2]
 105         raise ExternalCommandFailed(
 106             "Failed running external program: %s\nError: %s"
 107             % (cmd_string, "".join(traceback.format_exception_only(etype, value))))
 108     out, err = pipe.communicate()
 109     if "nothing changed" == out.strip(): # skip this error
 110         return out
 111     if (pipe.returncode != 0 or (fail_if_stderr and err.strip())) and not no_fail:
 112         raise ExternalCommandFailed(
 113             "External program failed (return code %d): %s\n%s\n%s"
 114             % (pipe.returncode, cmd_string, err, out))
 115     return out
 116
 117 def _run_raw_shell_command(cmd, no_fail=False):
 118     ui.status("* %s", cmd, level=ui.EXTRA, color='BLUE')
 119     st, out = commands.getstatusoutput(cmd)
 120     if st != 0 and not no_fail:
 121         raise ExternalCommandFailed(
 122             "External program failed with non-zero return code (%d): %s\n%s"
 123             % (st, cmd, out))
 124     return out
 125
 126 def run_command(cmd, args=None, bulk_args=None, encoding=None, fail_if_stderr=False, no_fail=False):
 127     """
 128     Run a command without using the shell.
 129     """
 130     args = args or []
 131     bulk_args = bulk_args or []
 132     def _transform_arg(a):
 133         if isinstance(a, unicode):
 134             a = a.encode(encoding or locale_encoding or 'UTF-8')
 135         elif not isinstance(a, str):
 136             a = str(a)
 137         return a
 138
 139     cmd = find_program(cmd)
 140     if not bulk_args:
 141         return _run_raw_command(cmd, map(_transform_arg, args), fail_if_stderr, no_fail)
 142     # If one of bulk_args starts with a dash (e.g. '-foo.php'),
 143     # svn will take this as an option. Adding '--' ends the search for
 144     # further options.
 145     for a in bulk_args:
 146         if a.strip().startswith('-'):
 147             args.append("--")
 148             break
 149     max_args_num = 254
 150     i = 0
 151     out = ""
 152     while i < len(bulk_args):
 153         stop = i + max_args_num - len(args)
 154         sub_args = []
 155         for a in bulk_args[i:stop]:
 156             sub_args.append(_transform_arg(a))
 157         out += _run_raw_command(cmd, args + sub_args, fail_if_stderr, no_fail)
 158         i = stop
 159     return out
 160
 161 def run_shell_command(cmd, args=None, bulk_args=None, encoding=None, no_fail=False):
 162     """
 163     Run a shell command, properly quoting and encoding arguments.
 164     Probably only works on Un*x-like systems.
 165     """
 166     def _quote_arg(a):
 167         if isinstance(a, unicode):
 168             a = a.encode(encoding or locale_encoding)
 169         elif not isinstance(a, str):
 170             a = str(a)
 171         return shell_quote(a)
 172
 173     if args:
 174         cmd += " " + " ".join(_quote_arg(a) for a in args)
 175     max_args_num = 254
 176     i = 0
 177     out = ""
 178     if not bulk_args:
 179         return _run_raw_shell_command(cmd, no_fail)
 180     while i < len(bulk_args):
 181         stop = i + max_args_num - len(args)
 182         sub_args = []
 183         for a in bulk_args[i:stop]:
 184             sub_args.append(_quote_arg(a))
 185         sub_cmd = cmd + " " + " ".join(sub_args)
 186         out += _run_raw_shell_command(sub_cmd, no_fail)
 187         i = stop
 188     return out
 189
 190 def run_svn(args=None, bulk_args=None, fail_if_stderr=False,
 191             mask_atsign=False, no_fail=False):
 192     """
 193     Run an SVN command, returns the (bytes) output.
 194     """
 195     if mask_atsign:
 196         # The @ sign in Subversion revers to a pegged revision number.
 197         # SVN treats files with @ in the filename a bit special.
 198         # See: http://stackoverflow.com/questions/1985203
 199         for idx in range(len(args)):
 200             if "@" in args[idx] and args[idx][0] not in ("-", '"'):
 201                 args[idx] = "%s@" % args[idx]
 202         if bulk_args:
 203             for idx in range(len(bulk_args)):
 204                 if ("@" in bulk_args[idx]
 205                     and bulk_args[idx][0] not in ("-", '"')):
 206                     bulk_args[idx] = "%s@" % bulk_args[idx]
 207     return run_command("svn",
 208         args=args, bulk_args=bulk_args, fail_if_stderr=fail_if_stderr, no_fail=no_fail)
 209
 210 def skip_dirs(paths, basedir="."):
 211     """
 212     Skip all directories from path list, including symbolic links to real dirs.
 213     """
 214     # NOTE: both tests are necessary (Cameron Hutchison's patch for symbolic
 215     # links to directories)
 216     return [p for p in paths
 217         if not os.path.isdir(os.path.join(basedir, p))
 218         or os.path.islink(os.path.join(basedir, p))]
 219
 220 def get_script_name():
 221     """Helper to return the name of the command line script that was called."""
 222     return os.path.basename(sys.argv[0])