data status: add --remote option to check files against a different remote (#10843)

dvc/commands/data.py

@@ -111,6 +111,7 @@ class CmdDataStatus(CmdBase):
                 targets=self.args.targets,
                 granular=self.args.granular,
                 untracked_files=self.args.untracked_files,
+                remote=self.args.remote,
                 not_in_remote=self.args.not_in_remote,
                 remote_refresh=self.args.remote_refresh,
             )
@@ -182,6 +183,12 @@ def add_parser(subparsers, parent_parser):
         nargs="?",
         help="Show untracked files.",
     )
+    data_status_parser.add_argument(
+        "-r",
+        "--remote",
+        help="Remote storage to check (only applicable with --not-in-remote).",
+        metavar="<name>",
+    )
     data_status_parser.add_argument(
         "--not-in-remote",
         action="store_true",

dvc/repo/data.py

@@ -404,13 +404,22 @@ def status(
     *,
     granular: bool = False,
     untracked_files: str = "no",
+    remote: Optional[str] = None,
     not_in_remote: bool = False,
     remote_refresh: bool = False,
+    config: Optional[dict] = None,
     batch_size: Optional[int] = None,
     head: str = "HEAD",
 ) -> Status:
     from dvc.scm import NoSCMError, SCMError
 
+    config = config or {}
+    if remote and not_in_remote:
+        logger.debug("Using remote %r", remote)
+        core = config.setdefault("core", {})
+        core["remote"] = remote
+    repo.config.merge(config)
+
     targets = targets or []
     filter_keys: list[DataIndexKey] = [repo.fs.relparts(os.fspath(t)) for t in targets]
     # try to remove duplicate and overlapping keys

tests/func/test_data_status.py

@@ -398,6 +398,42 @@ def test_missing_dir_object_from_index(M, tmp_dir, dvc, scm):
     }
 
 
+def test_remote_check(M, tmp_dir, dvc, scm, make_remote):
+    make_remote("default", default=True)
+    make_remote("myremote", default=False)
+
+    tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}})
+    tmp_dir.dvc_gen("foobar", "foobar")
+    assert dvc.push() == 4
+    scm.add_commit(["dir.dvc", "foobar.dvc", ".gitignore"], message="add files")
+
+    entries = M.unordered("foobar", join("dir", ""))
+    granular_entries = M.unordered(
+        "foobar", join("dir", ""), join("dir", "foo"), join("dir", "bar")
+    )
+    expected_ng = EMPTY_STATUS | {"git": M.dict(), "unchanged": entries}
+    expected_g = EMPTY_STATUS | {
+        "not_in_remote": [],
+        "git": M.dict(),
+        "unchanged": granular_entries,
+    }
+
+    opts = {"not_in_remote": True, "remote_refresh": True}
+    assert dvc.data_status(**opts) == expected_ng
+    assert dvc.data_status(granular=True, **opts) == expected_g
+
+    opts |= {"remote": "myremote"}
+    assert dvc.data_status(**opts) == expected_ng | {"not_in_remote": entries}
+    assert dvc.data_status(granular=True, **opts) == expected_g | {
+        "not_in_remote": granular_entries
+    }
+
+    dvc.push(remote="myremote")
+
+    assert dvc.data_status(**opts) == expected_ng
+    assert dvc.data_status(granular=True, **opts) == expected_g
+
+
 def test_missing_remote_cache(M, tmp_dir, dvc, scm, local_remote):
     tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}})
     tmp_dir.dvc_gen("foobar", "foobar")

tests/unit/command/test_data_status.py

@@ -43,6 +43,8 @@ def test_cli(dvc, mocker, mocked_status):
             "--unchanged",
             "--untracked-files",
             "--granular",
+            "--remote",
+            "myremote",
         ]
     )
 
@@ -53,6 +55,7 @@ def test_cli(dvc, mocker, mocked_status):
         targets=[],
         untracked_files="all",
         not_in_remote=False,
+        remote="myremote",
         remote_refresh=True,
         granular=True,
     )