diff --git a/python/runfiles/runfiles.py b/python/runfiles/runfiles.py index f7be6ef42f..03ebdf6331 100644 --- a/python/runfiles/runfiles.py +++ b/python/runfiles/runfiles.py @@ -29,7 +29,16 @@ import posixpath import sys from collections import defaultdict -from typing import Dict, List, Optional, Tuple, Union +from typing import Dict, Generator, Iterable, List, Optional, Tuple, Union + +if sys.version_info >= (3, 11): + from typing import Self +elif sys.version_info >= (3, 10): + from typing import TypeAlias + + Self: TypeAlias = "Path" # type: ignore +else: + from typing import Any as Self class _RepositoryMapping: @@ -143,25 +152,30 @@ def is_empty(self) -> bool: ) -class Path(pathlib.PurePath): +class Path(pathlib.Path): """A pathlib-like path object for runfiles. - This class extends `pathlib.PurePath` and resolves paths + This class extends `pathlib.Path` and resolves paths using the associated `Runfiles` instance when converted to a string. """ - # For Python < 3.12 compatibility when subclassing PurePath directly - _flavour = getattr(type(pathlib.PurePath()), "_flavour", None) + # Mypy isn't smart enough to realize `self` in the methods + # refers to our Path class instead of pathlib.Path + _runfiles: Optional["Runfiles"] + _source_repo: Optional[str] + + # For Python < 3.12 compatibility when subclassing Path directly + _flavour = getattr(type(pathlib.Path()), "_flavour", None) def __new__( cls, *args: Union[str, os.PathLike], runfiles: Optional["Runfiles"] = None, source_repo: Optional[str] = None, - ) -> "Path": + ) -> Self: """Private constructor. Use Runfiles.root() to create instances.""" obj = super().__new__(cls, *args) - # Type checkers might complain about adding attributes to PurePath, + # Type checkers might complain about adding attributes to Path, # but this is standard for pathlib subclasses. obj._runfiles = runfiles # type: ignore obj._source_repo = source_repo # type: ignore @@ -173,77 +187,188 @@ def __init__( runfiles: Optional["Runfiles"] = None, source_repo: Optional[str] = None, ) -> None: - pass + # In Python 3.12+, pathlib was refactored and Path.__init__ now accepts + # *args. Prior to 3.12, Path did not define __init__, so + # super().__init__(*args) would fall through to object.__init__, which + # raises a TypeError because it takes no arguments. + if sys.version_info >= (3, 12): + super().__init__(*args) + else: + super().__init__() + + # We override resolve() and absolute() to ensure that in Python < 3.12, + # where pathlib internally uses object.__new__ instead of our custom + # __new__ or with_segments(), the runfiles state is preserved. We delegate + # to self._as_path() because super().resolve() creates intermediate objects + # that would otherwise crash during internal stat() calls. + # override + def resolve(self, strict: bool = False) -> Self: + return type(self)( + self._as_path().resolve(strict=strict), + runfiles=self._runfiles, + source_repo=self._source_repo, + ) - def with_segments(self, *pathsegments: Union[str, os.PathLike]) -> "Path": + # override + def absolute(self) -> Self: + return type(self)( + self._as_path().absolute(), + runfiles=self._runfiles, + source_repo=self._source_repo, + ) + + # override + def with_segments(self, *pathsegments: Union[str, os.PathLike]) -> Self: """Used by Python 3.12+ pathlib to create new path objects.""" return type(self)( *pathsegments, - runfiles=self._runfiles, # type: ignore - source_repo=self._source_repo, # type: ignore + runfiles=self._runfiles, + source_repo=self._source_repo, ) # For Python < 3.12 - @classmethod - def _from_parts(cls, args: Tuple[str, ...]) -> "Path": - obj = super()._from_parts(args) # type: ignore - # These will be set by the calling instance later, or we can't set them here - # properly without context. Usually pathlib calls this from an instance - # method like _make_child, which we also might need to override. - return obj - - def _make_child(self, args: Tuple[str, ...]) -> "Path": + # override + def _make_child(self, args: Tuple[str, ...]) -> Self: obj = super()._make_child(args) # type: ignore obj._runfiles = self._runfiles # type: ignore obj._source_repo = self._source_repo # type: ignore return obj - @classmethod - def _from_parsed_parts(cls, drv: str, root: str, parts: List[str]) -> "Path": - obj = super()._from_parsed_parts(drv, root, parts) # type: ignore - return obj - - def _make_child_relpath(self, part: str) -> "Path": - obj = super()._make_child_relpath(part) # type: ignore - obj._runfiles = self._runfiles # type: ignore - obj._source_repo = self._source_repo # type: ignore - return obj - + # override @property - def parents(self) -> Tuple["Path", ...]: + def parents(self) -> Tuple[Self, ...]: return tuple( type(self)( p, - runfiles=getattr(self, "_runfiles", None), - source_repo=getattr(self, "_source_repo", None), + runfiles=self._runfiles, + source_repo=self._source_repo, ) for p in super().parents ) + # override @property - def parent(self) -> "Path": + def parent(self) -> Self: return type(self)( super().parent, - runfiles=getattr(self, "_runfiles", None), - source_repo=getattr(self, "_source_repo", None), + runfiles=self._runfiles, + source_repo=self._source_repo, ) - def with_name(self, name: str) -> "Path": + @property + def runfile_path(self) -> str: + """Returns the runfiles-root relative path.""" + path_posix = super().__str__().replace("\\", "/") + if path_posix == ".": + return "" + return path_posix + + # override + def with_name(self, name: str) -> Self: return type(self)( super().with_name(name), - runfiles=getattr(self, "_runfiles", None), - source_repo=getattr(self, "_source_repo", None), + runfiles=self._runfiles, + source_repo=self._source_repo, ) - def with_suffix(self, suffix: str) -> "Path": + # override + def with_suffix(self, suffix: str) -> Self: return type(self)( super().with_suffix(suffix), - runfiles=getattr(self, "_runfiles", None), - source_repo=getattr(self, "_source_repo", None), + runfiles=self._runfiles, + source_repo=self._source_repo, + ) + + def _as_path(self) -> pathlib.Path: + return pathlib.Path(str(self)) + + # override + def stat(self, *, follow_symlinks: bool = True) -> os.stat_result: + return self._as_path().stat(follow_symlinks=follow_symlinks) + + # override + def lstat(self) -> os.stat_result: + return self._as_path().lstat() + + # override + def exists(self) -> bool: + return self._as_path().exists() + + # override + def is_dir(self) -> bool: + return self._as_path().is_dir() + + # override + def is_file(self) -> bool: + return self._as_path().is_file() + + # override + def is_symlink(self) -> bool: + return self._as_path().is_symlink() + + # override + def is_block_device(self) -> bool: + return self._as_path().is_block_device() + + # override + def is_char_device(self) -> bool: + return self._as_path().is_char_device() + + # override + def is_fifo(self) -> bool: + return self._as_path().is_fifo() + + # override + def is_socket(self) -> bool: + return self._as_path().is_socket() + + # override + def open( + self, + mode: str = "r", + buffering: int = -1, + encoding: Optional[str] = None, + errors: Optional[str] = None, + newline: Optional[str] = None, + ): + return self._as_path().open( + mode=mode, + buffering=buffering, + encoding=encoding, + errors=errors, + newline=newline, ) + # override + def read_bytes(self) -> bytes: + return self._as_path().read_bytes() + + # override + def read_text( + self, encoding: Optional[str] = None, errors: Optional[str] = None + ) -> str: + return self._as_path().read_text(encoding=encoding, errors=errors) + + # override + def iterdir(self) -> Generator[Self, None, None]: + resolved = self._as_path() + for p in resolved.iterdir(): + yield self / p.name + + # override + def glob(self, pattern: str) -> Generator[Self, None, None]: + resolved = self._as_path() + for p in resolved.glob(pattern): + yield self / p.relative_to(resolved) + + # override + def rglob(self, pattern: str) -> Generator[Self, None, None]: + resolved = self._as_path() + for p in resolved.rglob(pattern): + yield self / p.relative_to(resolved) + def __repr__(self) -> str: - return 'runfiles.Path({!r})'.format(super().__str__()) + return "runfiles.Path({!r})".format(self.runfile_path) def __str__(self) -> str: path_posix = super().__str__().replace("\\", "/") @@ -251,12 +376,16 @@ def __str__(self) -> str: # pylint: disable=protected-access return self._runfiles._python_runfiles_root # type: ignore resolved = self._runfiles.Rlocation(path_posix, source_repo=self._source_repo) # type: ignore - return resolved if resolved is not None else super().__str__() + if resolved is not None: + return resolved + + # pylint: disable=protected-access + return posixpath.join(self._runfiles._python_runfiles_root, path_posix) # type: ignore def __fspath__(self) -> str: return str(self) - def runfiles_root(self) -> "Path": + def runfiles_root(self) -> Self: """Returns a Path object representing the runfiles root.""" return self._runfiles.root(source_repo=self._source_repo) # type: ignore diff --git a/tests/runfiles/BUILD.bazel b/tests/runfiles/BUILD.bazel index 7d675c7d7c..505b3c17c5 100644 --- a/tests/runfiles/BUILD.bazel +++ b/tests/runfiles/BUILD.bazel @@ -14,12 +14,34 @@ py_test( deps = ["//python/runfiles"], ) +py_test( + name = "runfiles_min_python_test", + srcs = ["runfiles_test.py"], + data = [ + "//tests/support:current_build_settings", + ], + env = { + "BZLMOD_ENABLED": "1" if BZLMOD_ENABLED else "0", + }, + main = "runfiles_test.py", + python_version = "3.10", + deps = ["//python/runfiles"], +) + py_test( name = "pathlib_test", srcs = ["pathlib_test.py"], deps = ["//python/runfiles"], ) +py_test( + name = "pathlib_min_python_test", + srcs = ["pathlib_test.py"], + main = "pathlib_test.py", + python_version = "3.10", + deps = ["//python/runfiles"], +) + build_test( name = "publishing", targets = [ diff --git a/tests/runfiles/pathlib_test.py b/tests/runfiles/pathlib_test.py index 553c8e4410..a959138235 100644 --- a/tests/runfiles/pathlib_test.py +++ b/tests/runfiles/pathlib_test.py @@ -9,15 +9,30 @@ class PathlibTest(unittest.TestCase): def setUp(self) -> None: self.tmpdir = tempfile.TemporaryDirectory(dir=os.environ.get("TEST_TMPDIR")) - # Runfiles paths are expected to be posix paths internally when we construct the strings for assertions - self.root_dir = pathlib.Path(self.tmpdir.name).as_posix() + # Runfiles paths are expected to be posix paths internally when we + # construct the strings for assertions + self.root_path = pathlib.Path(self.tmpdir.name).resolve() + self.root_dir = self.root_path.as_posix() + + # Create dummy files for I/O tests + self.repo_dir = self.root_path / "my_repo" + self.repo_dir.mkdir() + self.test_file = self.repo_dir / "data.txt" + self.test_file.write_text("hello runfiles", encoding="utf-8") + self.sub_dir = self.repo_dir / "subdir" + self.sub_dir.mkdir() + (self.sub_dir / "other.txt").write_text("other content", encoding="utf-8") + + def _create_runfiles(self) -> runfiles.Runfiles: + r = runfiles.Create({"RUNFILES_DIR": self.root_dir}) + assert r is not None + return r def tearDown(self) -> None: self.tmpdir.cleanup() def test_path_api(self) -> None: - r = runfiles.Create({"RUNFILES_DIR": self.root_dir}) - assert r is not None + r = self._create_runfiles() root = r.root() # Test basic joining @@ -41,26 +56,22 @@ def test_path_api(self) -> None: self.assertEqual(p, p3) def test_root(self) -> None: - r = runfiles.Create({"RUNFILES_DIR": self.root_dir}) - assert r is not None + r = self._create_runfiles() self.assertEqual(str(r.root()), self.root_dir) def test_runfiles_root_method(self) -> None: - r = runfiles.Create({"RUNFILES_DIR": self.root_dir}) - assert r is not None + r = self._create_runfiles() p = r.root() / "foo/bar" self.assertEqual(p.runfiles_root(), r.root()) self.assertEqual(str(p.runfiles_root()), self.root_dir) def test_os_path_like(self) -> None: - r = runfiles.Create({"RUNFILES_DIR": self.root_dir}) - assert r is not None + r = self._create_runfiles() p = r.root() / "foo" self.assertEqual(os.fspath(p), f"{self.root_dir}/foo") def test_equality_and_hash(self) -> None: - r = runfiles.Create({"RUNFILES_DIR": self.root_dir}) - assert r is not None + r = self._create_runfiles() p1 = r.root() / "foo" p2 = r.root() / "foo" p3 = r.root() / "bar" @@ -70,14 +81,12 @@ def test_equality_and_hash(self) -> None: self.assertEqual(hash(p1), hash(p2)) def test_join_path(self) -> None: - r = runfiles.Create({"RUNFILES_DIR": self.root_dir}) - assert r is not None + r = self._create_runfiles() p = r.root().joinpath("repo", "file") self.assertEqual(str(p), f"{self.root_dir}/repo/file") def test_parents(self) -> None: - r = runfiles.Create({"RUNFILES_DIR": self.root_dir}) - assert r is not None + r = self._create_runfiles() p = r.root() / "a/b/c" parents = list(p.parents) self.assertEqual(len(parents), 3) @@ -86,20 +95,99 @@ def test_parents(self) -> None: self.assertEqual(str(parents[2]), self.root_dir) def test_with_methods(self) -> None: - r = runfiles.Create({"RUNFILES_DIR": self.root_dir}) - assert r is not None + r = self._create_runfiles() p = r.root() / "foo/bar.txt" self.assertEqual(str(p.with_name("baz.py")), f"{self.root_dir}/foo/baz.py") self.assertEqual(str(p.with_suffix(".dat")), f"{self.root_dir}/foo/bar.dat") def test_match(self) -> None: - r = runfiles.Create({"RUNFILES_DIR": self.root_dir}) - assert r is not None + r = self._create_runfiles() p = r.root() / "foo/bar.txt" self.assertTrue(p.match("*.txt")) self.assertTrue(p.match("foo/*.txt")) self.assertFalse(p.match("bar/*.txt")) + def test_reading_api(self) -> None: + r = self._create_runfiles() + root = r.root() + p = root / "my_repo/data.txt" + + self.assertTrue(p.exists()) + self.assertTrue(p.is_file()) + self.assertEqual(p.read_text(encoding="utf-8"), "hello runfiles") + self.assertEqual(p.read_bytes(), b"hello runfiles") + + with p.open("r", encoding="utf-8") as f: + self.assertEqual(f.read(), "hello runfiles") + + def test_stat_api(self) -> None: + r = self._create_runfiles() + root = r.root() + p = root / "my_repo/data.txt" + + st = p.stat() + self.assertEqual(st.st_size, len("hello runfiles")) + + def test_iteration_api(self) -> None: + r = self._create_runfiles() + root = r.root() + p = root / "my_repo" + + self.assertTrue(p.is_dir()) + contents = {c.name for c in p.iterdir()} + self.assertEqual(contents, {"data.txt", "subdir"}) + # Ensure they are still runfiles.Path + for c in p.iterdir(): + self.assertIsInstance(c, runfiles.Path) + + def test_glob(self) -> None: + r = self._create_runfiles() + root = r.root() + p = root / "my_repo" + + glob_results = { + pathlib.PurePath(c).relative_to(pathlib.PurePath(p)).as_posix() + for c in p.glob("*.txt") + } + self.assertEqual(glob_results, {"data.txt"}) + + rglob_results = { + pathlib.PurePath(c).relative_to(pathlib.PurePath(p)).as_posix() + for c in p.rglob("*.txt") + } + self.assertEqual(rglob_results, {"data.txt", "subdir/other.txt"}) + + for c in p.rglob("*.txt"): + self.assertIsInstance(c, runfiles.Path) + + def test_resolve_and_absolute(self) -> None: + r = self._create_runfiles() + root = r.root() + p = root / "my_repo/data.txt" + + resolved = p.resolve() + self.assertIsInstance(resolved, runfiles.Path) + self.assertTrue(resolved.exists()) + self.assertEqual(resolved.read_text(encoding="utf-8"), "hello runfiles") + + absoluted = p.absolute() + self.assertIsInstance(absoluted, runfiles.Path) + self.assertTrue(absoluted.exists()) + self.assertEqual(absoluted.read_text(encoding="utf-8"), "hello runfiles") + + def test_runfile_path(self) -> None: + r = self._create_runfiles() + root = r.root() + p = root / "my_repo/data.txt" + self.assertEqual(p.runfile_path, "my_repo/data.txt") + + p2 = root / "foo" / "bar.txt" + self.assertEqual(p2.runfile_path, "foo/bar.txt") + + self.assertEqual(root.runfile_path, "") + self.assertEqual(repr(root), "runfiles.Path('')") + self.assertEqual(repr(p), "runfiles.Path('my_repo/data.txt')") + if __name__ == "__main__": unittest.main()