diff --git a/dvc/commands/data.py b/dvc/commands/data.py
index 31fd45079..b4b38c856 100644
--- a/dvc/commands/data.py
+++ b/dvc/commands/data.py
@@ -41,7 +41,8 @@ class CmdDataStatus(CmdBase):
         'use "git status" to see',
     }
 
-    def _process_status(self, status: "DataStatus"):
+    @staticmethod
+    def _process_status(status: "DataStatus"):
         """Flatten stage status, and filter empty stage status contents."""
         for stage, stage_status in status.items():
             items = stage_status
@@ -55,9 +56,10 @@ class CmdDataStatus(CmdBase):
                 continue
             yield stage, items
 
-    def _show_status(self, status: "DataStatus") -> int:
+    @classmethod
+    def _show_status(cls, status: "DataStatus") -> int:
         git_info = status.pop("git")  # type: ignore[misc]
-        result = dict(self._process_status(status))
+        result = dict(cls._process_status(status))
         if not result:
             no_changes = "No changes"
             if git_info.get("is_empty", False):
@@ -68,12 +70,12 @@ class CmdDataStatus(CmdBase):
             if idx:
                 ui.write()
 
-            label = self.LABELS.get(stage, stage.capitalize() + " files")
+            label = cls.LABELS.get(stage, stage.capitalize() + " files")
             header = f"{label}:"
-            color = self.COLORS.get(stage, "normal")
+            color = cls.COLORS.get(stage, "normal")
 
             ui.write(header)
-            if hint := self.HINTS.get(stage):
+            if hint := cls.HINTS.get(stage):
                 ui.write(f"  ({hint})")
 
             if isinstance(stage_status, dict):
@@ -87,7 +89,7 @@ class CmdDataStatus(CmdBase):
             for item in items:
                 ui.write(f"\t[{color}]{item}[/]".expandtabs(8), styled=True)
 
-        if (hint := self.HINTS.get("git_dirty")) and git_info.get("is_dirty"):
+        if (hint := cls.HINTS.get("git_dirty")) and git_info.get("is_dirty"):
             message = hint.format("other " if result else "")
             ui.write(f"[blue]({message})[/]", styled=True)
         return 0
@@ -102,6 +104,8 @@ class CmdDataStatus(CmdBase):
 
         if not self.args.unchanged:
             status.pop("unchanged")  # type: ignore[misc]
+        if self.args.untracked_files == "no":
+            status.pop("untracked")
         if self.args.json:
             status.pop("git")  # type: ignore[misc]
             ui.write_json(compact(status))
diff --git a/dvc/repo/data.py b/dvc/repo/data.py
index 38f6a1bc3..95cca2cf4 100644
--- a/dvc/repo/data.py
+++ b/dvc/repo/data.py
@@ -235,6 +235,7 @@ def status(repo: "Repo", untracked_files: str = "no", **kwargs: Any) -> Status:
         unchanged &= set(committed_diff.pop("unchanged", []))
 
     git_info = _git_info(repo.scm, untracked_files=untracked_files)
+    # order matters here
     return Status(
         not_in_cache=not_in_cache,
         committed=committed_diff,
diff --git a/setup.cfg b/setup.cfg
index 07da27adc..249f31d91 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -125,6 +125,7 @@ tests =
     pytest-xdist==2.5.0
     pytest-mock==3.8.2
     pytest-lazy-fixture==0.6.3
+    pytest-test-utils==0.0.8
     # https://github.com/docker/docker-py/issues/2902
     pytest-docker==0.11.0; python_version < '3.10' or sys_platform != 'win32'
     flaky==3.7.0
diff --git a/tests/func/test_data_status.py b/tests/func/test_data_status.py
new file mode 100644
index 000000000..77ef8d7b6
--- /dev/null
+++ b/tests/func/test_data_status.py
@@ -0,0 +1,150 @@
+from os.path import join
+
+from dvc.repo import Repo
+from dvc.utils.fs import remove
+
+EMPTY_STATUS = {
+    "committed": {},
+    "uncommitted": {},
+    "git": {},
+    "not_in_cache": [],
+    "unchanged": [],
+    "untracked": [],
+}
+
+
+def test_file(matcher, tmp_dir, scm, dvc):
+    tmp_dir.dvc_gen("foo", "foo", commit="add foo")
+    tmp_dir.dvc_gen("foo", "foobar")
+    remove(tmp_dir / "foo")
+
+    expected = {
+        "committed": {"modified": ["foo"]},
+        "uncommitted": {"deleted": ["foo"]},
+        "git": matcher.instance_of(dict),
+        "not_in_cache": [],
+        "unchanged": [],
+        "untracked": [],
+    }
+    assert dvc.data_status() == expected
+    assert dvc.data_status(granular=True) == expected
+
+
+def test_directory(matcher, tmp_dir, scm, dvc):
+    tmp_dir.dvc_gen({"dir": {"foo": "foo"}}, commit="add dir")
+    tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar", "foobar": "foobar"}})
+    remove(tmp_dir / "dir")
+    (tmp_dir / "dir").gen({"foo": "foo", "bar": "barr", "baz": "baz"})
+    tmp_dir.gen("untracked", "untracked")
+
+    assert dvc.data_status() == {
+        "committed": {"modified": [join("dir", "")]},
+        "uncommitted": {"modified": [join("dir", "")]},
+        "git": matcher.instance_of(dict),
+        "not_in_cache": [],
+        "unchanged": [],
+        "untracked": [],
+    }
+
+    assert dvc.data_status(granular=True, untracked_files="all") == {
+        "committed": {
+            "added": matcher.unordered(
+                join("dir", "bar"),
+                join("dir", "foobar"),
+            )
+        },
+        "uncommitted": {
+            "added": [join("dir", "baz")],
+            "modified": [join("dir", "bar")],
+            "deleted": [join("dir", "foobar")],
+        },
+        "git": matcher.instance_of(dict),
+        "not_in_cache": [],
+        "unchanged": [join("dir", "foo")],
+        "untracked": ["untracked"],
+    }
+
+
+def test_new_empty_git_repo(matcher, tmp_dir, scm):
+    dvc = Repo.init()
+    assert dvc.data_status() == {
+        **EMPTY_STATUS,
+        "git": matcher.dict(
+            is_empty=True,
+            is_dirty=True,
+        ),
+    }
+
+
+def test_noscm_repo(dvc):
+    assert dvc.data_status() == EMPTY_STATUS
+
+
+def test_unchanged(matcher, tmp_dir, scm, dvc):
+    tmp_dir.dvc_gen({"dir": {"foo": "foo"}}, commit="add dir")
+    tmp_dir.dvc_gen("bar", "bar", commit="add foo")
+
+    assert dvc.data_status() == {
+        **EMPTY_STATUS,
+        "git": matcher.instance_of(dict),
+        "unchanged": matcher.unordered("bar", join("dir", "")),
+    }
+    assert dvc.data_status(granular=True) == {
+        **EMPTY_STATUS,
+        "git": matcher.instance_of(dict),
+        "unchanged": matcher.unordered("bar", join("dir", "foo")),
+    }
+
+
+def test_not_in_cache(matcher, tmp_dir, scm, dvc):
+    # TODO: investigation required, might return wrong results
+    tmp_dir.dvc_gen({"dir": {"foo": "foo"}}, commit="add dir")
+    tmp_dir.dvc_gen("bar", "bar", commit="add foo")
+    remove(dvc.odb.local.cache_dir)
+
+    assert dvc.data_status() == {
+        **EMPTY_STATUS,
+        "not_in_cache": matcher.unordered("bar", join("dir", "")),
+        "git": matcher.instance_of(dict),
+        "unchanged": ["bar"],
+        "uncommitted": {"added": [join("dir", "")]},
+    }
+    assert dvc.data_status(granular=True) == {
+        **EMPTY_STATUS,
+        "not_in_cache": matcher.unordered("bar"),
+        "git": matcher.instance_of(dict),
+        "unchanged": [],
+        "committed": {"modified": ["bar"]},
+        "uncommitted": {"modified": ["bar"], "added": [join("dir", "foo")]},
+    }
+
+
+def test_withdirs(matcher, tmp_dir, scm, dvc):
+    tmp_dir.dvc_gen({"dir": {"foo": "foo"}}, commit="add dir")
+    tmp_dir.dvc_gen("bar", "bar", commit="add foo")
+    assert dvc.data_status(granular=True, with_dirs=True) == {
+        **EMPTY_STATUS,
+        "git": matcher.instance_of(dict),
+        "unchanged": matcher.unordered(
+            "bar", join("dir", "foo"), join("dir", "")
+        ),
+    }
+
+
+def test_skip_uncached_pipeline_outputs(tmp_dir, dvc, run_copy_metrics):
+    tmp_dir.gen({"m_temp.yaml": str(5)})
+    run_copy_metrics(
+        "m_temp.yaml",
+        "m.yaml",
+        metrics_no_cache=["m.yaml"],
+        name="copy-metrics",
+    )
+    assert dvc.data_status() == EMPTY_STATUS
+    assert (
+        dvc.data_status(granular=True, untracked_files="all") == EMPTY_STATUS
+    )
+
+
+def test_output_with_newly_added_stage(tmp_dir, dvc):
+    dvc.stage.add(deps=["bar"], outs=["foo"], name="copy", cmd="cp foo bar")
+    assert dvc.data_status() == {**EMPTY_STATUS, "not_in_cache": ["foo"]}
diff --git a/tests/unit/command/test_data_status.py b/tests/unit/command/test_data_status.py
new file mode 100644
index 000000000..eb6109091
--- /dev/null
+++ b/tests/unit/command/test_data_status.py
@@ -0,0 +1,139 @@
+import json
+
+import pytest
+from funcy import omit
+
+from dvc.cli import main, parse_args
+from dvc.commands.data import CmdDataStatus
+from dvc.repo import Repo
+from dvc.repo.data import Status
+
+
+@pytest.fixture
+def mocked_status():
+    yield Status(
+        not_in_cache=["notincache"],
+        committed={"added": ["dir/bar", "dir/foo"]},
+        uncommitted={
+            "added": ["dir/baz"],
+            "modified": ["dir/bar"],
+            "deleted": ["dir/foobar"],
+        },
+        untracked=["untracked"],
+        unchanged=["dir/foo"],
+        git={"is_dirty": True},
+    )
+
+
+def test_cli(dvc, mocker, mocked_status):
+    status = mocker.patch(
+        "dvc.repo.Repo.data_status", return_value=mocked_status
+    )
+
+    cli_args = parse_args(
+        [
+            "data",
+            "status",
+            "--json",
+            "--unchanged",
+            "--untracked-files",
+            "--with-dirs",
+            "--granular",
+        ]
+    )
+
+    assert cli_args.func == CmdDataStatus
+    cmd = cli_args.func(cli_args)
+    assert cmd.run() == 0
+    status.assert_called_once_with(
+        untracked_files="all",
+        granular=True,
+        with_dirs=True,
+    )
+
+
+@pytest.mark.parametrize(
+    "args, to_omit",
+    [
+        ([], ["untracked", "unchanged"]),
+        (["--unchanged"], ["untracked"]),
+        (["--unchanged", "--untracked-files"], []),
+    ],
+)
+def test_json(dvc, mocker, capsys, mocked_status, args, to_omit):
+    mocker.patch("dvc.repo.Repo.data_status", return_value=mocked_status)
+    assert main(["data", "status", "--json", *args]) == 0
+    out, err = capsys.readouterr()
+    assert out.rstrip() == json.dumps(omit(mocked_status, [*to_omit, "git"]))
+    assert not err
+
+
+def test_no_changes_repo(dvc, scm, capsys):
+    assert main(["data", "status"]) == 0
+    assert capsys.readouterr() == ("No changes.\n", "")
+
+
+def test_empty_scm_repo(tmp_dir, capsys):
+    tmp_dir.init(scm=True)
+    Repo.init()
+
+    assert main(["data", "status"]) == 0
+    out, err = capsys.readouterr()
+    assert (
+        out
+        == """\
+No changes in an empty git repo.
+(there are changes not tracked by dvc, use "git status" to see)
+"""
+    )
+    assert not err
+
+
+@pytest.mark.parametrize(
+    "args",
+    [
+        ("--untracked-files",),
+        ("--unchanged",),
+        ("--untracked-files", "--unchanged"),
+    ],
+)
+@pytest.mark.parametrize("is_dirty", [True, False])
+def test_show_status(dvc, scm, mocker, capsys, mocked_status, args, is_dirty):
+    mocked_status["git"]["is_dirty"] = is_dirty
+    mocker.patch("dvc.repo.Repo.data_status", return_value=mocked_status)
+    assert main(["data", "status", *args]) == 0
+    out, err = capsys.readouterr()
+    expected_out = """\
+Not in cache:
+  (use "dvc pull <file>..." to update your local storage)
+        notincache
+
+DVC committed changes:
+  (git commit the corresponding dvc files to update the repo)
+        added: dir/bar
+        added: dir/foo
+
+DVC uncommitted changes:
+  (use "dvc commit <file>..." to track changes)
+        added: dir/baz
+        modified: dir/bar
+        deleted: dir/foobar
+"""
+    if "--untracked-files" in args:
+        expected_out += """
+Untracked files:
+  (use "git add <file> ..." or dvc add <file>..." to commit to git or to dvc)
+        untracked
+"""
+    if "--unchanged" in args:
+        expected_out += """
+DVC unchanged files:
+        dir/foo
+"""
+
+    if is_dirty:
+        expected_out += """\
+(there are other changes not tracked by dvc, use "git status" to see)
+"""
+    assert out == expected_out
+    assert not err