svn2svn/shell.py

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