dvc: return non zero error code on push/pull/fetch error

Closes #2038

dvc/command/data_sync.py

@@ -4,6 +4,7 @@ import argparse
 import logging
 
 from dvc.command.base import CmdBase, append_doc_link
+from dvc.exceptions import DvcException
 
 
 logger = logging.getLogger(__name__)
@@ -45,7 +46,7 @@ class CmdDataPull(CmdDataBase):
                 force=self.args.force,
                 recursive=self.args.recursive,
             )
-        except Exception:
+        except DvcException:
             logger.exception("failed to pull data from the cloud")
             return 1
         self.check_up_to_date(processed_files_count)
@@ -65,7 +66,7 @@ class CmdDataPush(CmdDataBase):
                 with_deps=self.args.with_deps,
                 recursive=self.args.recursive,
             )
-        except Exception:
+        except DvcException:
             logger.exception("failed to push data to the cloud")
             return 1
         self.check_up_to_date(processed_files_count)
@@ -85,7 +86,7 @@ class CmdDataFetch(CmdDataBase):
                 with_deps=self.args.with_deps,
                 recursive=self.args.recursive,
             )
-        except Exception:
+        except DvcException:
             logger.exception("failed to fetch data from the cloud")
             return 1
         self.check_up_to_date(processed_files_count)

dvc/remote/base.py

@@ -390,6 +390,7 @@ class RemoteBASE(object):
         if not hasattr(self, "_upload"):
             raise RemoteActionNotImplemented("upload", self.scheme)
         names = self._verify_path_args(to_infos, from_infos, names)
+        fails = 0
 
         with self.transfer_context() as ctx:
             for from_info, to_info, name in zip(from_infos, to_infos, names):
@@ -417,6 +418,7 @@ class RemoteBASE(object):
                         no_progress_bar=no_progress_bar,
                     )
                 except Exception:
+                    fails += 1
                     msg = "failed to upload '{}' to '{}'"
                     logger.exception(msg.format(from_info, to_info))
                     continue
@@ -424,18 +426,21 @@ class RemoteBASE(object):
                 if not no_progress_bar:
                     progress.finish_target(name)
 
+        return fails
+
     def download(
         self,
         from_infos,
         to_infos,
-        no_progress_bar=False,
         names=None,
+        no_progress_bar=False,
         resume=False,
     ):
         if not hasattr(self, "_download"):
             raise RemoteActionNotImplemented("download", self.scheme)
 
         names = self._verify_path_args(from_infos, to_infos, names)
+        fails = 0
 
         with self.transfer_context() as ctx:
             for to_info, from_info, name in zip(to_infos, from_infos, names):
@@ -473,6 +478,7 @@ class RemoteBASE(object):
                         no_progress_bar=no_progress_bar,
                     )
                 except Exception:
+                    fails += 1
                     msg = "failed to download '{}' to '{}'"
                     logger.exception(msg.format(from_info, to_info))
                     continue
@@ -482,6 +488,8 @@ class RemoteBASE(object):
                 if not no_progress_bar:
                     progress.finish_target(name)
 
+        return fails
+
     def remove(self, path_info):
         raise RemoteActionNotImplemented("remove", self.scheme)
 

dvc/remote/http.py

@@ -48,11 +48,12 @@ class RemoteHTTP(RemoteBASE):
         self,
         from_infos,
         to_infos,
-        no_progress_bar=False,
         names=None,
+        no_progress_bar=False,
         resume=False,
     ):
         names = self._verify_path_args(to_infos, from_infos, names)
+        fails = 0
 
         for to_info, from_info, name in zip(to_infos, from_infos, names):
             if from_info.scheme != self.scheme:
@@ -82,6 +83,7 @@ class RemoteHTTP(RemoteBASE):
                 )
 
             except Exception:
+                fails += 1
                 msg = "failed to download '{}'".format(from_info)
                 logger.exception(msg)
                 continue
@@ -89,6 +91,8 @@ class RemoteHTTP(RemoteBASE):
             if not no_progress_bar:
                 progress.finish_target(name)
 
+        return fails
+
     def exists(self, path_infos):
         single_path = False
 

dvc/remote/local/__init__.py

@@ -355,26 +355,16 @@ class RemoteLOCAL(RemoteBASE):
             info["status"] = status
 
     def _get_chunks(self, download, remote, status_info, status, jobs):
-        title = "Analysing status."
-
-        progress.set_n_total(1)
-        total = len(status_info)
-        current = 0
-
         cache = []
         path_infos = []
         names = []
-        for md5, info in status_info.items():
+        for md5, info in progress(
+            status_info.items(), name="Analysing status"
+        ):
             if info["status"] == status:
                 cache.append(self.checksum_to_path_info(md5))
                 path_infos.append(remote.checksum_to_path_info(md5))
                 names.append(info["name"])
-            current += 1
-            progress.update_target(title, current, total)
-
-        progress.finish_target(title)
-
-        progress.set_n_total(len(names))
 
         if download:
             to_infos = cache
@@ -383,12 +373,10 @@ class RemoteLOCAL(RemoteBASE):
             to_infos = path_infos
             from_infos = cache
 
-        return list(
-            zip(
-                to_chunks(from_infos, jobs),
-                to_chunks(to_infos, jobs),
-                to_chunks(names, jobs),
-            )
+        return (
+            to_chunks(from_infos, jobs),
+            to_chunks(to_infos, jobs),
+            to_chunks(names, jobs),
         )
 
     def _process(
@@ -399,11 +387,9 @@ class RemoteLOCAL(RemoteBASE):
         show_checksums=False,
         download=False,
     ):
-        msg = "Preparing to {} data {} '{}'"
         logger.info(
-            msg.format(
-                "download" if download else "upload",
-                "from" if download else "to",
+            "Preparing to {} '{}'".format(
+                "download data from" if download else "upload data to",
                 remote.path_info,
             )
         )
@@ -428,25 +414,22 @@ class RemoteLOCAL(RemoteBASE):
 
         chunks = self._get_chunks(download, remote, status_info, status, jobs)
 
-        if len(chunks) == 0:
+        if len(chunks[0]) == 0:
             return 0
 
         if jobs > 1:
-            futures = []
             with ThreadPoolExecutor(max_workers=jobs) as executor:
-                for from_infos, to_infos, names in chunks:
-                    res = executor.submit(
-                        func, from_infos, to_infos, names=names
-                    )
-                    futures.append(res)
-
-            for f in futures:
-                f.result()
+                fails = sum(executor.map(func, *chunks))
         else:
-            for from_infos, to_infos, names in chunks:
-                func(from_infos, to_infos, names=names)
+            fails = sum(map(func, *chunks))
 
-        return len(chunks)
+        if fails:
+            msg = "{} file(s) failed to {}"
+            raise DvcException(
+                msg.format(fails, "download" if download else "upload")
+            )
+
+        return len(chunks[0])
 
     def push(self, checksum_infos, remote, jobs=None, show_checksums=False):
         return self._process(

tests/func/test_remote.py

@@ -1,14 +1,17 @@
 import os
 import configobj
+import shutil
 from mock import patch
 
 from dvc.main import main
 from dvc.config import Config
 from dvc.remote.base import RemoteBASE
+from dvc.remote import RemoteLOCAL
 from dvc.path_info import PathInfo
 
 from tests.basic_env import TestDvc
 from tests.func.test_data_cloud import get_local_storagepath
+from .test_data_cloud import get_local_url
 
 
 class TestRemote(TestDvc):
@@ -205,3 +208,41 @@ def test_dir_checksum_should_be_key_order_agnostic(dvc_repo):
             checksum2 = dvc_repo.cache.local.get_dir_checksum(path_info)
 
     assert checksum1 == checksum2
+
+
+def test_partial_push_n_pull(dvc_repo, repo_dir, caplog):
+    assert main(["remote", "add", "-d", "upstream", get_local_url()]) == 0
+    # Recreate the repo to reread config
+    repo = dvc_repo.__class__(dvc_repo.root_dir)
+    remote = repo.cloud.get_remote("upstream")
+
+    foo = repo.add(repo_dir.FOO)[0].outs[0]
+    bar = repo.add(repo_dir.BAR)[0].outs[0]
+
+    # Faulty upload version, failing on foo
+    original = RemoteLOCAL._upload
+
+    def unreliable_upload(self, from_file, to_info, name=None, **kwargs):
+        if name == "foo":
+            raise Exception("stop foo")
+        return original(self, from_file, to_info, name, **kwargs)
+
+    with patch.object(RemoteLOCAL, "_upload", unreliable_upload):
+        assert main(["push"]) == 1
+        assert str(get_last_exc(caplog)) == "1 file(s) failed to upload"
+
+        assert not remote.exists(remote.checksum_to_path_info(foo.checksum))
+        assert remote.exists(remote.checksum_to_path_info(bar.checksum))
+
+    # Push everything and delete local cache
+    assert main(["push"]) == 0
+    shutil.rmtree(repo.cache.local.cache_dir)
+
+    with patch.object(RemoteLOCAL, "_download", side_effect=Exception):
+        assert main(["pull"]) == 1
+        assert str(get_last_exc(caplog)) == "2 file(s) failed to download"
+
+
+def get_last_exc(caplog):
+    _, exc, _ = caplog.records[-1].exc_info
+    return exc