repro: Move pull logic to inside Stage.

dvc/repo/reproduce.py

@@ -194,10 +194,6 @@ def _reproduce_stages(  # noqa: C901
             )
 
         try:
-            if kwargs.get("pull") and stage.changed():
-                logger.debug("Pulling %s", stage.addressing)
-                stage.repo.pull(stage.addressing, allow_missing=True)
-
             ret = _reproduce_stage(stage, **kwargs)
 
             if len(ret) == 0:

dvc/stage/__init__.py

@@ -561,7 +561,7 @@ def commit(self, allow_missing=False, filter_info=None, **kwargs) -> None:
             raise CacheLinkError(link_failures)
 
     @rwlocked(read=["deps", "outs"])
-    def run(
+    def run(  # noqa: C901
         self,
         dry=False,
         no_commit=False,
@@ -573,15 +573,21 @@ def run(
         if (self.cmd or self.is_import) and not self.frozen and not dry:
             self.remove_outs(ignore_remove=False, force=False)
 
-        if (not self.frozen and self.is_import) or self.is_partial_import:
+        if (
+            self.is_import and (not self.frozen or kwargs.get("pull"))
+        ) or self.is_partial_import:
             self._sync_import(dry, force, kwargs.get("jobs", None), no_download)
         elif not self.frozen and self.cmd:
             self._run_stage(dry, force, **kwargs)
-        else:
+        elif kwargs.get("pull"):
+            logger.info("Pulling data for %s", self)
+            for objs in self.get_used_objs().values():
+                self.repo.cloud.pull(objs)
+            self.checkout()
+        elif not dry:
             args = ("outputs", "frozen ") if self.frozen else ("data sources", "")
             logger.info("Verifying %s in %s%s", *args, self)
-            if not dry:
-                self._check_missing_outputs()
+            self._check_missing_outputs()
 
         if not dry:
             if kwargs.get("checkpoint_func", None) or no_download: