svn2svn/shell.py

   1 """ Shell functions """
   2
   3 from svn2svn import ui
   4 from svn2svn.errors import ExternalCommandFailed
   5
   6 import os
   7 import locale
   8 from datetime import datetime
   9 import time
  10 from subprocess import Popen, PIPE, STDOUT
  11 import shutil
  12 import stat
  13 import sys
  14 import traceback
  15 import re
  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 locale_encoding = locale.getpreferredencoding()
  76
  77 def get_encoding():
  78     return locale_encoding
  79
  80 def shell_quote(s):
  81     global _debug_showcmd
  82     if _debug_showcmd:
  83         # If showing OS commands being executed, don't wrap "safe" strings in quotes.
  84         if re.compile('^[A-Za-z0-9=-]+$').match(s):
  85             return s
  86     if os.name == "nt":
  87         q = '"'
  88     else:
  89         q = "'"
  90     return q + s.replace('\\', '\\\\').replace("'", "'\"'\"'") + q
  91
  92 def _run_raw_command(cmd, args, fail_if_stderr=False):
  93     cmd_string = "%s %s" % (cmd,  " ".join(map(shell_quote, args)))
  94     ui.status("* %s", cmd_string, level=ui.DEBUG)
  95     try:
  96         pipe = Popen([cmd] + args, executable=cmd, stdout=PIPE, stderr=PIPE)
  97     except OSError:
  98         etype, value = sys.exc_info()[:2]
  99         raise ExternalCommandFailed(
 100             "Failed running external program: %s\nError: %s"
 101             % (cmd_string, "".join(traceback.format_exception_only(etype, value))))
 102     out, err = pipe.communicate()
 103     if "nothing changed" == out.strip(): # skip this error
 104         return out
 105     if pipe.returncode != 0 or (fail_if_stderr and err.strip()):
 106         raise ExternalCommandFailed(
 107             "External program failed (return code %d): %s\n%s\n%s"
 108             % (pipe.returncode, cmd_string, err, out))
 109     return out
 110
 111 def _run_raw_shell_command(cmd):
 112     ui.status("* %s", cmd, level=ui.DEBUG)
 113     st, out = commands.getstatusoutput(cmd)
 114     if st != 0:
 115         raise ExternalCommandFailed(
 116             "External program failed with non-zero return code (%d): %s\n%s"
 117             % (st, cmd, out))
 118     return out
 119
 120 def run_command(cmd, args=None, bulk_args=None, encoding=None, fail_if_stderr=False):
 121     """
 122     Run a command without using the shell.
 123     """
 124     args = args or []
 125     bulk_args = bulk_args or []
 126     def _transform_arg(a):
 127         if isinstance(a, unicode):
 128             a = a.encode(encoding or locale_encoding or 'UTF-8')
 129         elif not isinstance(a, str):
 130             a = str(a)
 131         return a
 132
 133     cmd = find_program(cmd)
 134     if not bulk_args:
 135         return _run_raw_command(cmd, map(_transform_arg, args), fail_if_stderr)
 136     # If one of bulk_args starts with a dash (e.g. '-foo.php'),
 137     # svn will take this as an option. Adding '--' ends the search for
 138     # further options.
 139     for a in bulk_args:
 140         if a.strip().startswith('-'):
 141             args.append("--")
 142             break
 143     max_args_num = 254
 144     i = 0
 145     out = ""
 146     while i < len(bulk_args):
 147         stop = i + max_args_num - len(args)
 148         sub_args = []
 149         for a in bulk_args[i:stop]:
 150             sub_args.append(_transform_arg(a))
 151         out += _run_raw_command(cmd, args + sub_args, fail_if_stderr)
 152         i = stop
 153     return out
 154
 155 def run_shell_command(cmd, args=None, bulk_args=None, encoding=None):
 156     """
 157     Run a shell command, properly quoting and encoding arguments.
 158     Probably only works on Un*x-like systems.
 159     """
 160     def _quote_arg(a):
 161         if isinstance(a, unicode):
 162             a = a.encode(encoding or locale_encoding)
 163         elif not isinstance(a, str):
 164             a = str(a)
 165         return shell_quote(a)
 166
 167     if args:
 168         cmd += " " + " ".join(_quote_arg(a) for a in args)
 169     max_args_num = 254
 170     i = 0
 171     out = ""
 172     if not bulk_args:
 173         return _run_raw_shell_command(cmd)
 174     while i < len(bulk_args):
 175         stop = i + max_args_num - len(args)
 176         sub_args = []
 177         for a in bulk_args[i:stop]:
 178             sub_args.append(_quote_arg(a))
 179         sub_cmd = cmd + " " + " ".join(sub_args)
 180         out += _run_raw_shell_command(sub_cmd)
 181         i = stop
 182     return out
 183
 184 def run_svn(args=None, bulk_args=None, fail_if_stderr=False,
 185             mask_atsign=False):
 186     """
 187     Run an SVN command, returns the (bytes) output.
 188     """
 189     if mask_atsign:
 190         # The @ sign in Subversion revers to a pegged revision number.
 191         # SVN treats files with @ in the filename a bit special.
 192         # See: http://stackoverflow.com/questions/1985203
 193         for idx in range(len(args)):
 194             if "@" in args[idx] and args[idx][0] not in ("-", '"'):
 195                 args[idx] = "%s@" % args[idx]
 196         if bulk_args:
 197             for idx in range(len(bulk_args)):
 198                 if ("@" in bulk_args[idx]
 199                     and bulk_args[idx][0] not in ("-", '"')):
 200                     bulk_args[idx] = "%s@" % bulk_args[idx]
 201     return run_command("svn",
 202         args=args, bulk_args=bulk_args, fail_if_stderr=fail_if_stderr)
 203
 204 def skip_dirs(paths, basedir="."):
 205     """
 206     Skip all directories from path list, including symbolic links to real dirs.
 207     """
 208     # NOTE: both tests are necessary (Cameron Hutchison's patch for symbolic
 209     # links to directories)
 210     return [p for p in paths
 211         if not os.path.isdir(os.path.join(basedir, p))
 212         or os.path.islink(os.path.join(basedir, p))]
 213
 214 def get_script_name():
 215     """Helper to return the name of the command line script that was called."""
 216     return os.path.basename(sys.argv[0])