gitfs: add regression test for broken stage.collect on gitfs when cwd != repo_root

tests/func/test_api.py

@@ -4,7 +4,11 @@ import pytest
 from funcy import first, get_in
 
 from dvc import api
-from dvc.exceptions import FileMissingError, OutputNotFoundError
+from dvc.exceptions import (
+    FileMissingError,
+    OutputNotFoundError,
+    PathMissingError,
+)
 from dvc.testing.test_api import TestAPI  # noqa, pylint: disable=unused-import
 from dvc.utils.fs import remove
 from tests.unit.fs.test_repo import make_subrepo
@@ -207,7 +211,10 @@ def test_get_url_subrepos(tmp_dir, scm, local_cloud):
     assert api.get_url("subrepo/bar") == expected_url
 
 
-@pytest.mark.xfail(reason="https://github.com/iterative/dvc/issues/7341")
+@pytest.mark.xfail(
+    raises=PathMissingError,
+    reason="https://github.com/iterative/dvc/issues/7341",
+)
 def test_open_from_remote(tmp_dir, erepo_dir, cloud, local_cloud):
     erepo_dir.add_remote(config=cloud.config, name="other")
     erepo_dir.add_remote(config=local_cloud.config, default=True)
@@ -216,6 +223,6 @@ def test_open_from_remote(tmp_dir, erepo_dir, cloud, local_cloud):
     remove(erepo_dir.dvc.odb.local.cache_dir)
 
     with api.open(
-        os.path.join("dir", "foo"), repo=erepo_dir.as_uri(), remote="other"
+        os.path.join("dir", "foo"), repo=f"file://{erepo_dir}", remote="other"
     ) as fd:
         assert fd.read() == "foo content"

tests/func/test_used_objs.py

@@ -0,0 +1,63 @@
+import json
+import os
+
+import pytest
+
+from dvc.exceptions import NoOutputOrStageError
+from dvc.stage.exceptions import StageFileDoesNotExistError
+
+gitfs_xfail = pytest.mark.xfail(
+    raises=(NoOutputOrStageError, StageFileDoesNotExistError),
+    reason="gitfs works on absolute paths only",
+)
+
+
+@pytest.mark.parametrize(
+    "stage_wdir, cwd, target",
+    [
+        (os.curdir, os.curdir, "foo"),
+        (os.curdir, os.curdir, "train"),
+        (os.curdir, os.curdir, "dvc.yaml:train"),
+        (os.curdir, "sub", os.path.join(os.pardir, "foo")),
+        pytest.param(
+            os.curdir,
+            "sub",
+            os.path.join(os.pardir, "dvc.yaml:train"),
+            marks=gitfs_xfail,
+        ),
+        ("sub", os.curdir, os.path.join("sub", "foo")),
+        ("sub", os.curdir, os.path.join("sub", "dvc.yaml:train")),
+        ("sub", "sub", "foo"),
+        pytest.param("sub", "sub", "train", marks=gitfs_xfail),
+        pytest.param("sub", "sub", "dvc.yaml:train", marks=gitfs_xfail),
+        ("sub", "dir", os.path.join(os.pardir, "sub", "foo")),
+        pytest.param(
+            "sub",
+            "dir",
+            os.path.join(os.pardir, "sub", "dvc.yaml:train"),
+            marks=gitfs_xfail,
+        ),
+    ],
+)
+def test_from_gitfs_when_pwd_not_in_root(
+    tmp_dir, scm, dvc, stage_wdir, cwd, target
+):
+    path = tmp_dir.joinpath(stage_wdir).resolve()
+    path.mkdir(parents=True, exist_ok=True)
+    wdir = tmp_dir.joinpath(cwd).resolve()
+    wdir.mkdir(parents=True, exist_ok=True)
+
+    (path / "dvc.yaml").write_text(
+        json.dumps(
+            {"stages": {"train": {"cmd": "echo foo > foo", "outs": ["foo"]}}}
+        )
+    )
+    path.gen({"foo": "foo"})
+    dvc.commit(None, force=True)
+    tmp_dir.scm_add(
+        [path / file for file in ("dvc.yaml", "dvc.lock", ".gitignore")],
+        commit="add files",
+    )
+
+    with wdir.chdir():
+        assert dvc.used_objs([target], revs=[scm.get_rev()])