Merge pull request #900 from efiop/master

remote: misc bug fixes

dvc/dependency/base.py

@@ -68,5 +68,5 @@ class DependencyBase(object):
     def dumpd(self):
         return {self.PARAM_PATH: self.path}
 
-    def download(self, path):
-        self.remote.download([self.path_info], [path])
+    def download(self, to_info):
+        self.remote.download([self.path_info], [to_info])

dvc/remote/gs.py

@@ -55,7 +55,7 @@ class RemoteGS(RemoteBase):
 
         return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], path_info['key'])}
 
-    def _copy(from_info, to_info, gs=None):
+    def _copy(self, from_info, to_info, gs=None):
         gs = gs if gs else self.gs
  
         blob = gs.bucket(from_info['bucket']).get_blob(from_info['key'])
@@ -160,7 +160,7 @@ class RemoteGS(RemoteBase):
             if from_info['scheme'] != 'gs':
                 raise NotImplementedError
 
-            if from_info['scheme'] == 'gs':
+            if to_info['scheme'] == 'gs':
                 self._copy(from_info, to_info, gs=gs)
                 continue
 

dvc/remote/local.py

@@ -369,12 +369,18 @@ class RemoteLOCAL(RemoteBase):
         if len(chunks) == 0:
             return
 
+        futures = []
         with ThreadPoolExecutor(max_workers=len(chunks)) as executor:
             for from_infos, to_infos, md5s in chunks:
-                executor.submit(remote.download, from_infos,
-                                                 to_infos,
-                                                 names=md5s,
-                                                 no_progress_bar=no_progress_bar)
+                res = executor.submit(remote.download,
+                                      from_infos,
+                                      to_infos,
+                                      names=md5s,
+                                      no_progress_bar=no_progress_bar)
+                futures.append(res)
+
+        for f in futures:
+            f.result()
 
     def pull(self, checksum_infos, remote, jobs=1):
         # NOTE: try fetching missing dir info
@@ -410,6 +416,11 @@ class RemoteLOCAL(RemoteBase):
         if len(chunks) == 0:
             return
 
+        futures = []
         with ThreadPoolExecutor(max_workers=len(chunks)) as executor:
             for to_infos, from_infos, md5s in chunks:
-                executor.submit(remote.upload, from_infos, to_infos, names=md5s)
+                res = executor.submit(remote.upload, from_infos, to_infos, names=md5s)
+                futures.append(res)
+
+        for f in futures:
+            f.result()

dvc/remote/s3.py

@@ -167,7 +167,7 @@ class RemoteS3(RemoteBase):
 
         return ret
 
-    def upload(self, from_infos, path_infos, names=None):
+    def upload(self, from_infos, to_infos, names=None):
         names = self._verify_path_args(to_infos, from_infos, names)
 
         s3 = self.s3_session_client