Outputs as target supporting for dvc status (#4433)

* Outputs as target supporting for `dvc status`

fix #4191
1. Add a related test which would fail on current version.

* Pass the tests

1. add deps to the tests.
2. make change to pass the tests.

* Update tests/func/test_status.py

Co-authored-by: Saugat Pachhai <[email protected]>

* Solve some change request.

Co-authored-by: Saugat Pachhai <[email protected]>

Author: karajan1001

dvc/repo/status.py

@@ -10,31 +10,32 @@
 logger = logging.getLogger(__name__)
 
 
-def _joint_status(stages):
+def _joint_status(pairs):
     status_info = {}
 
-    for stage in stages:
+    for stage, filter_info in pairs:
         if stage.frozen and not stage.is_repo_import:
             logger.warning(
                 "{} is frozen. Its dependencies are"
                 " not going to be shown in the status output.".format(stage)
             )
-
-        status_info.update(stage.status(check_updates=True))
+        if not filter_info:
+            status_info.update(stage.status(check_updates=True))
+        else:
+            for out in stage.filter_outs(filter_info):
+                status_info.update(out.status())
 
     return status_info
 
 
 def _local_status(self, targets=None, with_deps=False, recursive=False):
-    if targets:
-        stages = cat(
-            self.collect(t, with_deps=with_deps, recursive=recursive)
-            for t in targets
-        )
-    else:
-        stages = self.collect(None, with_deps=with_deps, recursive=recursive)
+    targets = targets or [None]
+    pairs = cat(
+        self.collect_granular(t, with_deps=with_deps, recursive=recursive)
+        for t in targets
+    )
 
-    return _joint_status(stages)
+    return _joint_status(pairs)
 
 
 def _cloud_status(

dvc/stage/__init__.py

@@ -442,7 +442,7 @@ def run(self, dry=False, no_commit=False, force=False, run_cache=True):
             if not no_commit:
                 self.commit()
 
-    def _filter_outs(self, path_info):
+    def filter_outs(self, path_info):
         def _func(o):
             return path_info.isin_or_eq(o.path_info)
 
@@ -451,7 +451,7 @@ def _func(o):
     @rwlocked(write=["outs"])
     def checkout(self, **kwargs):
         stats = defaultdict(list)
-        for out in self._filter_outs(kwargs.get("filter_info")):
+        for out in self.filter_outs(kwargs.get("filter_info")):
             key, outs = self._checkout(out, **kwargs)
             if key:
                 stats[key].extend(outs)
@@ -526,14 +526,14 @@ def outs_cached(self):
     def get_all_files_number(self, filter_info=None):
         return sum(
             out.get_files_number(filter_info)
-            for out in self._filter_outs(filter_info)
+            for out in self.filter_outs(filter_info)
         )
 
     def get_used_cache(self, *args, **kwargs):
         from dvc.cache import NamedCache
 
         cache = NamedCache()
-        for out in self._filter_outs(kwargs.get("filter_info")):
+        for out in self.filter_outs(kwargs.get("filter_info")):
             cache.update(out.get_used_cache(*args, **kwargs))
 
         return cache

tests/func/test_status.py

@@ -118,3 +118,22 @@ def test_status_recursive(tmp_dir, dvc):
             }
         ],
     }
+
+
+def test_status_outputs(tmp_dir, dvc):
+    tmp_dir.dvc_gen({"foo": "foo", "bar": "bar"})
+    dvc.run(
+        outs=["alice", "bob"],
+        deps=["foo", "bar"],
+        cmd="echo alice>alice && echo bob>bob",
+        name="alice_bob",
+    )
+    tmp_dir.gen({"alice": "new alice", "bob": "new bob"})
+
+    assert dvc.status(targets=["alice_bob"]) == {
+        "alice_bob": [
+            {"changed outs": {"alice": "modified", "bob": "modified"}}
+        ]
+    }
+
+    assert dvc.status(targets=["alice"]) == {"alice": "modified"}