remote, #525: pump fetch-infos instead of GIL-read stderr

+ `handle_process_output()` accepts null-finalizer, to pump completely
stderr before raising any errors.
+ test: Enable `TestGit.test_environment()` on Windows (to checks stderr
consumption).

Author: ankostis

git/cmd.py

@@ -59,7 +59,8 @@
 # Documentation
 ## @{
 
-def handle_process_output(process, stdout_handler, stderr_handler, finalizer, decode_streams=True):
+def handle_process_output(process, stdout_handler, stderr_handler,
+                          finalizer=None, decode_streams=True):
     """Registers for notifications to lean that process output is ready to read, and dispatches lines to
     the respective line handlers.
     This function returns once the finalizer returns
@@ -108,10 +109,13 @@ def pump_stream(cmdline, name, stream, is_decode, handler):
         t.start()
         threads.append(t)
 
+    ## FIXME: Why Join??  Will block if `stdin` needs feeding...
+    #
     for t in threads:
         t.join()
 
-    return finalizer(process)
+    if finalizer:
+        return finalizer(process)
 
 
 def dashify(string):

git/remote.py

@@ -630,25 +630,19 @@ def _get_fetch_info_from_stderr(self, proc, progress):
         cmds = set(PushInfo._flag_map.keys()) & set(FetchInfo._flag_map.keys())
 
         progress_handler = progress.new_message_handler()
+        handle_process_output(proc, None, progress_handler, finalizer=None, decode_streams=False)
 
-        stderr_text = None
+        stderr_text = progress.error_lines and '\n'.join(progress.error_lines) or ''
+        proc.wait(stderr=stderr_text)
+        if stderr_text:
+            log.warning("Error lines received while fetching: %s", stderr_text)
 
-        for line in proc.stderr:
+        for line in progress.other_lines:
             line = force_text(line)
-            for pline in progress_handler(line):
-                # END handle special messages
-                for cmd in cmds:
-                    if len(line) > 1 and line[0] == ' ' and line[1] == cmd:
-                        fetch_info_lines.append(line)
-                        continue
-                    # end find command code
-                # end for each comand code we know
-            # end for each line progress didn't handle
-        # end
-        if progress.error_lines():
-            stderr_text = '\n'.join(progress.error_lines())
-
-        finalize_process(proc, stderr=stderr_text)
+            for cmd in cmds:
+                if len(line) > 1 and line[0] == ' ' and line[1] == cmd:
+                    fetch_info_lines.append(line)
+                    continue
 
         # read head information
         with open(join(self.repo.git_dir, 'FETCH_HEAD'), 'rb') as fp:

git/test/test_git.py

@@ -207,18 +207,15 @@ def test_environment(self, rw_dir):
         rw_repo = Repo.init(os.path.join(rw_dir, 'repo'))
         remote = rw_repo.create_remote('ssh-origin', "ssh://git@server/foo")
 
-        # This only works if we are not evaluating git-push/pull output in a thread !
-        import select
-        if hasattr(select, 'poll'):
-            with rw_repo.git.custom_environment(GIT_SSH=path):
-                try:
-                    remote.fetch()
-                except GitCommandError as err:
-                    if sys.version_info[0] < 3 and is_darwin:
-                        self.assertIn('ssh-orig, ' in str(err))
-                        self.assertEqual(err.status, 128)
-                    else:
-                        self.assertIn('FOO', str(err))
+        with rw_repo.git.custom_environment(GIT_SSH=path):
+            try:
+                remote.fetch()
+            except GitCommandError as err:
+                if sys.version_info[0] < 3 and is_darwin:
+                    self.assertIn('ssh-orig, ' in str(err))
+                    self.assertEqual(err.status, 128)
+                else:
+                    self.assertIn('FOO', str(err))
 
     def test_handle_process_output(self):
         from git.cmd import handle_process_output

git/util.py

@@ -199,33 +199,34 @@ class RemoteProgress(object):
     DONE_TOKEN = 'done.'
     TOKEN_SEPARATOR = ', '
 
-    __slots__ = ("_cur_line", "_seen_ops", "_error_lines")
+    __slots__ = ('_cur_line',
+                 '_seen_ops',
+                 'error_lines',  # Lines that started with 'error:' or 'fatal:'.
+                 'other_lines')  # Lines not denoting progress (i.e.g. push-infos).
     re_op_absolute = re.compile(r"(remote: )?([\w\s]+):\s+()(\d+)()(.*)")
     re_op_relative = re.compile(r"(remote: )?([\w\s]+):\s+(\d+)% \((\d+)/(\d+)\)(.*)")
 
     def __init__(self):
         self._seen_ops = list()
         self._cur_line = None
-        self._error_lines = []
-
-    def error_lines(self):
-        """Returns all lines that started with error: or fatal:"""
-        return self._error_lines
+        self.error_lines = []
+        self.other_lines = []
 
     def _parse_progress_line(self, line):
         """Parse progress information from the given line as retrieved by git-push
         or git-fetch.
 
-        Lines that seem to contain an error (i.e. start with error: or fatal:) are stored
-        separately and can be queried using `error_lines()`.
+        - Lines that do not contain progress info are stored in :attr:`other_lines`.
+        - Lines that seem to contain an error (i.e. start with error: or fatal:) are stored
+        in :attr:`error_lines`.
 
         :return: list(line, ...) list of lines that could not be processed"""
         # handle
         # Counting objects: 4, done.
         # Compressing objects:  50% (1/2)   \rCompressing objects: 100% (2/2)   \rCompressing objects: 100% (2/2), done.
         self._cur_line = line
-        if len(self._error_lines) > 0 or self._cur_line.startswith(('error:', 'fatal:')):
-            self._error_lines.append(self._cur_line)
+        if len(self.error_lines) > 0 or self._cur_line.startswith(('error:', 'fatal:')):
+            self.error_lines.append(self._cur_line)
             return []
 
         sub_lines = line.split('\r')
@@ -284,6 +285,7 @@ def _parse_progress_line(self, line):
                 self.line_dropped(sline)
                 # Note: Don't add this line to the failed lines, as we have to silently
                 # drop it
+                self.other_lines.extend(failed_lines)
                 return failed_lines
             # END handle op code
 
@@ -309,6 +311,7 @@ def _parse_progress_line(self, line):
                         max_count and float(max_count),
                         message)
         # END for each sub line
+        self.other_lines.extend(failed_lines)
         return failed_lines
 
     def new_message_handler(self):