cygwin, #525: Try to make it work with Cygwin's Git.

+ Make `Git.polish_url()` convert paths into Cygwin-friendly paths.
+ Add utility and soe TCs for funcs for detecting cygwin and converting
abs-paths to `/cygdrive/c/...`.

Author: ankostis

git/cmd.py

@@ -31,6 +31,7 @@
 )
 from git.exc import CommandError
 from git.odict import OrderedDict
+from git.util import is_cygwin_git, cygpath
 
 from .exc import (
     GitCommandError,
@@ -190,9 +191,24 @@ def __setstate__(self, d):
     # Override this value using `Git.USE_SHELL = True`
     USE_SHELL = False
 
+    @classmethod
+    def is_cygwin(cls):
+        return is_cygwin_git(cls.GIT_PYTHON_GIT_EXECUTABLE)
+
     @classmethod
     def polish_url(cls, url):
-        return url.replace("\\\\", "\\").replace("\\", "/")
+        if cls.is_cygwin():
+            """Remove any backslahes from urls to be written in config files.
+
+            Windows might create config-files containing paths with backslashed,
+            but git stops liking them as it will escape the backslashes.
+            Hence we undo the escaping just to be sure.
+            """
+            url = cygpath(url)
+        else:
+            url = url.replace("\\\\", "\\").replace("\\", "/")
+
+        return url
 
     class AutoInterrupt(object):
         """Kill/Interrupt the stored process instance once this instance goes out of scope. It is

git/repo/base.py

@@ -4,51 +4,21 @@
 # This module is part of GitPython and is released under
 # the BSD License: http://www.opensource.org/licenses/bsd-license.php
 
-from git.exc import (
-    InvalidGitRepositoryError,
-    NoSuchPathError,
-    GitCommandError
-)
-from git.cmd import (
-    Git,
-    handle_process_output
-)
-from git.refs import (
-    HEAD,
-    Head,
-    Reference,
-    TagReference,
-)
-from git.objects import (
-    Submodule,
-    RootModule,
-    Commit
-)
-from git.util import (
-    Actor,
-    finalize_process
-)
-from git.index import IndexFile
-from git.config import GitConfigParser
-from git.remote import (
-    Remote,
-    add_progress,
-    to_progress_instance
-)
-
-from git.db import GitCmdObjectDB
+from collections import namedtuple
+import logging
+import os
+import re
+import sys
 
 from gitdb.util import (
     join,
     isfile,
     hex_to_bin
 )
 
-from .fun import (
-    rev_parse,
-    is_git_dir,
-    find_git_dir,
-    touch,
+from git.cmd import (
+    Git,
+    handle_process_output
 )
 from git.compat import (
     text_type,
@@ -58,12 +28,17 @@
     range,
     is_win,
 )
+from git.config import GitConfigParser
+from git.db import GitCmdObjectDB
+from git.exc import InvalidGitRepositoryError, NoSuchPathError, GitCommandError
+from git.index import IndexFile
+from git.objects import Submodule, RootModule, Commit
+from git.refs import HEAD, Head, Reference, TagReference
+from git.remote import Remote, add_progress, to_progress_instance
+from git.util import Actor, finalize_process
+
+from .fun import rev_parse, is_git_dir, find_git_dir, touch
 
-import os
-import sys
-import re
-import logging
-from collections import namedtuple
 
 log = logging.getLogger(__name__)
 
@@ -875,12 +850,22 @@ def _clone(cls, git, url, path, odb_default_type, progress, **kwargs):
             progress = to_progress_instance(progress)
 
         odbt = kwargs.pop('odbt', odb_default_type)
-        proc = git.clone(url, path, with_extended_output=True, as_process=True,
+
+        ## A bug win cygwin's Git, when `--bare`
+        #  it prepends the basename of the `url` into the `path::
+        #        git clone --bare  /cygwin/a/foo.git  C:\\Work
+        #  becomes::
+        #        git clone --bare  /cygwin/a/foo.git  /cygwin/a/C:\\Work
+        #
+        clone_path = (Git.polish_url(path)
+                      if Git.is_cygwin() and 'bare' in kwargs
+                      else path)
+        proc = git.clone(Git.polish_url(url), clone_path, with_extended_output=True, as_process=True,
                          v=True, **add_progress(kwargs, git, progress))
         if progress:
             handle_process_output(proc, None, progress.new_message_handler(), finalize_process)
         else:
-            (stdout, stderr) = proc.communicate()  # FIXME: Will block of outputs are big!
+            (stdout, stderr) = proc.communicate()  # FIXME: Will block if outputs are big!
             log.debug("Cmd(%s)'s unused stdout: %s", getattr(proc, 'args', ''), stdout)
             finalize_process(proc, stderr=stderr)
 

git/test/lib/helper.py

@@ -32,7 +32,7 @@
     'GIT_REPO', 'GIT_DAEMON_PORT'
 )
 
-log = logging.getLogger('git.util')
+log = logging.getLogger(__name__)
 
 #{ Routines
 

git/test/test_util.py

@@ -5,7 +5,19 @@
 # the BSD License: http://www.opensource.org/licenses/bsd-license.php
 
 import tempfile
+import time
+from unittest.case import skipIf
+
+import ddt
 
+from git.cmd import dashify
+from git.compat import string_types, is_win
+from git.objects.util import (
+    altz_to_utctz_str,
+    utctz_to_altz,
+    verify_utctz,
+    parse_date,
+)
 from git.test.lib import (
     TestBase,
     assert_equal
@@ -15,19 +27,9 @@
     BlockingLockFile,
     get_user_id,
     Actor,
-    IterableList
+    IterableList,
+    cygpath,
 )
-from git.objects.util import (
-    altz_to_utctz_str,
-    utctz_to_altz,
-    verify_utctz,
-    parse_date,
-)
-from git.cmd import dashify
-from git.compat import string_types, is_win
-
-import time
-import ddt
 
 
 class TestIterableMember(object):
@@ -52,6 +54,47 @@ def setup(self):
             "array": [42],
         }
 
+    @skipIf(not is_win, "Paths specifically for Windows.")
+    @ddt.data(
+        (r'foo\bar', 'foo/bar'),
+        (r'foo/bar', 'foo/bar'),
+        (r'./bar', 'bar'),
+        (r'.\bar', 'bar'),
+        (r'../bar', '../bar'),
+        (r'..\bar', '../bar'),
+        (r'../bar/.\foo/../chu', '../bar/chu'),
+
+        (r'C:\Users', '/cygdrive/c/Users'),
+        (r'C:\d/e', '/cygdrive/c/d/e'),
+
+        (r'\\?\a:\com', '/cygdrive/a/com'),
+        (r'\\?\a:/com', '/cygdrive/a/com'),
+
+        (r'\\server\C$\Users', '//server/C$/Users'),
+        (r'\\server\C$', '//server/C$'),
+        (r'\\server\BAR/', '//server/BAR/'),
+        (r'\\?\UNC\server\D$\Apps', '//server/D$/Apps'),
+
+        (r'D:/Apps', '/cygdrive/d/Apps'),
+        (r'D:/Apps\fOO', '/cygdrive/d/Apps/fOO'),
+        (r'D:\Apps/123', '/cygdrive/d/Apps/123'),
+    )
+    def test_cygpath_ok(self, case):
+        wpath, cpath = case
+        self.assertEqual(cygpath(wpath), cpath or wpath)
+
+    @skipIf(not is_win, "Paths specifically for Windows.")
+    @ddt.data(
+        (r'C:Relative', None),
+        (r'D:Apps\123', None),
+        (r'D:Apps/123', None),
+        (r'\\?\a:rel', None),
+        (r'\\share\a:rel', None),
+    )
+    def test_cygpath_invalids(self, case):
+        wpath, cpath = case
+        self.assertEqual(cygpath(wpath), cpath or wpath.replace('\\', '/'))
+
     def test_it_should_dashify(self):
         assert_equal('this-is-my-argument', dashify('this_is_my_argument'))
         assert_equal('foo', dashify('foo'))

git/util.py

@@ -5,6 +5,8 @@
 # the BSD License: http://www.opensource.org/licenses/bsd-license.php
 from __future__ import unicode_literals
 
+import contextlib
+from functools import wraps
 import getpass
 import logging
 import os
@@ -13,10 +15,8 @@
 import shutil
 import stat
 import time
+from unittest.case import SkipTest
 
-from functools import wraps
-
-from git.compat import is_win
 from gitdb.util import (# NOQA @IgnorePep8
     make_sha,
     LockedFD,               # @UnusedImport
@@ -26,6 +26,7 @@
     to_bin_sha              # @UnusedImport
 )
 
+from git.compat import is_win
 import os.path as osp
 
 from .compat import (
@@ -34,7 +35,6 @@
     PY3
 )
 from .exc import InvalidGitRepositoryError
-from unittest.case import SkipTest
 
 
 # NOTE:  Some of the unused imports might be used/imported by others.
@@ -47,6 +47,8 @@
            'RemoteProgress', 'CallableRemoteProgress', 'rmtree', 'unbare_repo',
            'HIDE_WINDOWS_KNOWN_ERRORS')
 
+log = logging.getLogger(__name__)
+
 #: We need an easy way to see if Appveyor TCs start failing,
 #: so the errors marked with this var are considered "acknowledged" ones, awaiting remedy,
 #: till then, we wish to hide them.
@@ -70,6 +72,16 @@ def wrapper(self, *args, **kwargs):
     return wrapper
 
 
+@contextlib.contextmanager
+def cwd(new_dir):
+    old_dir = os.getcwd()
+    os.chdir(new_dir)
+    try:
+        yield new_dir
+    finally:
+        os.chdir(old_dir)
+
+
 def rmtree(path):
     """Remove the given recursively.
 
@@ -162,14 +174,141 @@ def assure_directory_exists(path, is_file=False):
         Otherwise it must be a directory
     :return: True if the directory was created, False if it already existed"""
     if is_file:
-        path = os.path.dirname(path)
+        path = osp.dirname(path)
     # END handle file
-    if not os.path.isdir(path):
+    if not osp.isdir(path):
         os.makedirs(path)
         return True
     return False
 
 
+def _get_exe_extensions():
+    try:
+        winprog_exts = tuple(p.upper() for p in os.environ['PATHEXT'].split(os.pathsep))
+    except:
+        winprog_exts = ('.BAT', 'COM', '.EXE')
+
+    return winprog_exts
+
+
+def py_where(program, path=None):
+    # From: http://stackoverflow.com/a/377028/548792
+    try:
+        winprog_exts = tuple(p.upper() for p in os.environ['PATHEXT'].split(os.pathsep))
+    except:
+        winprog_exts = is_win and ('.BAT', 'COM', '.EXE') or ()
+
+    def is_exec(fpath):
+        return osp.isfile(fpath) and os.access(fpath, os.X_OK) and (
+            os.name != 'nt' or not winprog_exts or any(fpath.upper().endswith(ext)
+                                                       for ext in winprog_exts))
+
+    progs = []
+    if not path:
+        path = os.environ["PATH"]
+    for folder in path.split(osp.pathsep):
+        folder = folder.strip('"')
+        if folder:
+            exe_path = osp.join(folder, program)
+            for f in [exe_path] + ['%s%s' % (exe_path, e) for e in winprog_exts]:
+                if is_exec(f):
+                    progs.append(f)
+    return progs
+
+
+def _cygexpath(drive, path):
+    if osp.isabs(path) and not drive:
+        ## Invoked from `cygpath()` directly with `D:Apps\123`?
+        #  It's an error, leave it alone just slashes)
+        p = path
+    else:
+        p = osp.normpath(osp.expandvars(os.path.expanduser(path)))
+        if osp.isabs(p):
+            if drive:
+                # Confusing, maybe a remote system should expand vars.
+                p = path
+            else:
+                p = cygpath(p)
+        elif drive:
+            p = '/cygdrive/%s/%s' % (drive.lower(), p)
+
+    return p.replace('\\', '/')
+
+
+_cygpath_parsers = (
+    ## See: https://msdn.microsoft.com/en-us/library/windows/desktop/aa365247(v=vs.85).aspx
+    ## and: https://www.cygwin.com/cygwin-ug-net/using.html#unc-paths
+    (re.compile(r"\\\\\?\\UNC\\([^\\]+)\\([^\\]+)(?:\\(.*))?"),
+     (lambda server, share, rest_path: '//%s/%s/%s' % (server, share, rest_path.replace('\\', '/'))),
+     False
+     ),
+
+    (re.compile(r"\\\\\?\\(\w):[/\\](.*)"),
+     _cygexpath,
+     False
+     ),
+
+    (re.compile(r"(\w):[/\\](.*)"),
+     _cygexpath,
+     False
+     ),
+
+    (re.compile(r"file:(.*)", re.I),
+     (lambda rest_path: rest_path),
+     True),
+
+    (re.compile(r"(\w{2,}:.*)"),  # remote URL, do nothing
+     (lambda url: url),
+     False),
+)
+
+
+def cygpath(path):
+    if not path.startswith(('/cygdrive', '//')):
+        for regex, parser, recurse in _cygpath_parsers:
+            match = regex.match(path)
+            if match:
+                path = parser(*match.groups())
+                if recurse:
+                    path = cygpath(path)
+                break
+        else:
+            path = _cygexpath(None, path)
+
+    return path
+
+
+#: Store boolean flags denoting if a specific Git executable
+#: is from a Cygwin installation (since `cache_lru()` unsupported on PY2).
+_is_cygwin_cache = {}
+
+
+def is_cygwin_git(git_executable):
+    if not is_win:
+        return False
+
+    from subprocess import check_output
+
+    is_cygwin = _is_cygwin_cache.get(git_executable)
+    if is_cygwin is None:
+        is_cygwin = False
+        try:
+            git_dir = osp.dirname(git_executable)
+            if not git_dir:
+                res = py_where(git_executable)
+                git_dir = osp.dirname(res[0]) if res else None
+
+            ## Just a name given, not a real path.
+            uname_cmd = osp.join(git_dir, 'uname')
+            uname = check_output(uname_cmd, universal_newlines=True)
+            is_cygwin = 'CYGWIN' in uname
+        except Exception as ex:
+            log.debug('Failed checking if running in CYGWIN due to: %r', ex)
+        _is_cygwin_cache[git_executable] = is_cygwin
+
+    return is_cygwin
+
+
 def get_user_id():
     """:return: string identifying the currently active system user as name@node"""
     return "%s@%s" % (getpass.getuser(), platform.node())
@@ -589,7 +728,7 @@ def _obtain_lock_or_raise(self):
         if self._has_lock():
             return
         lock_file = self._lock_file_path()
-        if os.path.isfile(lock_file):
+        if osp.isfile(lock_file):
             raise IOError("Lock for file %r did already exist, delete %r in case the lock is illegal" %
                           (self._file_path, lock_file))
 
@@ -659,7 +798,7 @@ def _obtain_lock(self):
                 # synity check: if the directory leading to the lockfile is not
                 # readable anymore, raise an execption
                 curtime = time.time()
-                if not os.path.isdir(os.path.dirname(self._lock_file_path())):
+                if not osp.isdir(osp.dirname(self._lock_file_path())):
                     msg = "Directory containing the lockfile %r was not readable anymore after waiting %g seconds" % (
                         self._lock_file_path(), curtime - starttime)
                     raise IOError(msg)