--jobs now takes effect on imported data (#4765)

* `--jobs` now takes effect on imported data

fix #4692

* Tests for this change

* Solve issue on python3.6 and 3.7

* Update tests/unit/test_external_repo.py

* Update tests/unit/test_external_repo.py

* Update tests/unit/test_external_repo.py

* Update tests/unit/test_external_repo.py

* tests: fix formatting

Co-authored-by: Ruslan Kuprieiev <[email protected]>
Co-authored-by: Ruslan Kuprieiev <[email protected]>

dvc/external_repo.py

@@ -130,13 +130,15 @@ class ExternalRepo(Repo):
             return self.scm.get_rev()
         return self.tree.rev
 
-    def _fetch_to_cache(self, path_info, repo, callback):
+    def _fetch_to_cache(self, path_info, repo, callback, **kwargs):
         # don't support subrepo traversal as it might fail due to difference
         # in remotes
         tree = self._get_tree_for(repo)
         cache = repo.cache.local
 
-        hash_info = tree.get_hash(path_info, download_callback=callback)
+        hash_info = tree.get_hash(
+            path_info, download_callback=callback, **kwargs
+        )
         cache.save(
             path_info,
             tree,
@@ -170,7 +172,9 @@ class ExternalRepo(Repo):
 
             self._check_repo(path, metadata.repo)
             repo = metadata.repo
-            hash_info = self._fetch_to_cache(path, repo, download_update)
+            hash_info = self._fetch_to_cache(
+                path, repo, download_update, **kwargs
+            )
             hash_infos.append(hash_info)
 
         return sum(download_results), failed, hash_infos

tests/unit/test_external_repo.py

@@ -77,3 +77,25 @@ def test_subrepo_is_constructed_properly(
             assert repo.config["remote"]["auto-generated-upstream"][
                 "url"
             ] == str(main_cache)
+
+
+def test_fetch_external_repo_jobs(tmp_dir, scm, mocker, dvc, local_remote):
+    tmp_dir.dvc_gen(
+        {
+            "dir1": {
+                "file1": "file1",
+                "file2": "file2",
+                "file3": "file3",
+                "file4": "file4",
+            },
+        },
+        commit="init",
+    )
+
+    dvc.push()
+
+    with external_repo(str(tmp_dir)) as repo:
+        spy = mocker.spy(repo.cloud, "pull")
+        repo.fetch_external(["dir1"], jobs=3)
+        run_jobs = tuple(spy.call_args_list[0])[1].get("jobs")
+        assert run_jobs == 3