Path.relative_to() walk_up support

.pre-commit-config.yaml

@@ -12,7 +12,7 @@ repos:
       - id: debug-statements
       - id: end-of-file-fixer
       - id: mixed-line-ending
-        args: [ "--fix=lf" ]
+        args: ["--fix=lf"]
       - id: trailing-whitespace
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
@@ -26,6 +26,7 @@ repos:
     rev: v1.8.0
     hooks:
       - id: mypy
+        args: [--no-warn-unused-ignores]
         additional_dependencies:
           - pytest
           - trio >= 0.23
@@ -34,6 +35,6 @@ repos:
   - repo: https://github.com/pre-commit/pygrep-hooks
     rev: v1.10.0
     hooks:
-    - id: rst-backticks
-    - id: rst-directive-colons
-    - id: rst-inline-touching-normal
+      - id: rst-backticks
+      - id: rst-directive-colons
+      - id: rst-inline-touching-normal

docs/versionhistory.rst

@@ -3,6 +3,14 @@ Version history
 
 This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 
+**UNRELEASED**
+
+- Added support for the python 3.12 ``walk_up`` keyword argument in
+  ``anyio.Path.relative_to()`` (PR by Colin Taylor)
+- Modified socket tests to generate unix socket paths of lengths less than 107 bytes.
+  Modern unices enforce this limitation and throw an OSError when the path is too long
+  (PR by Colin Taylor)
+
 **4.2.0**
 
 - Add support for ``byte``-based paths in ``connect_unix``, ``create_unix_listeners``,

src/anyio/_core/_fileio.py

@@ -514,8 +514,17 @@ async def read_text(
     ) -> str:
         return await to_thread.run_sync(self._path.read_text, encoding, errors)
 
-    def relative_to(self, *other: str | PathLike[str]) -> Path:
-        return Path(self._path.relative_to(*other))
+    if sys.version_info >= (3, 12):
+
+        def relative_to(
+            self, *other: str | PathLike[str], walk_up: bool = False
+        ) -> Path:
+            return Path(self._path.relative_to(*other, walk_up=walk_up))
+
+    else:
+
+        def relative_to(self, *other: str | PathLike[str]) -> Path:
+            return Path(self._path.relative_to(*other))
 
     async def readlink(self) -> Path:
         target = await to_thread.run_sync(os.readlink, self._path)

tests/test_fileio.py

@@ -1,16 +1,18 @@
 from __future__ import annotations
 
+import contextlib
 import os
 import pathlib
 import platform
 import socket
 import stat
 import sys
+from types import TracebackType
 
 import pytest
 from _pytest.tmpdir import TempPathFactory
 
-from anyio import AsyncFile, Path, open_file, wrap_file
+from anyio import AsyncFile, Path, open_file, to_thread, wrap_file
 
 pytestmark = pytest.mark.anyio
 
@@ -64,14 +66,40 @@ async def test_wrap_file(self, tmp_path: pathlib.Path) -> None:
 
 
 class TestPath:
+    class ChangeDirectory(contextlib.AbstractAsyncContextManager):
+        def __init__(self, cwd: Path) -> None:
+            super().__init__()
+            self.__cwd: Path = cwd
+            self.__pwd: Path | None = None
+
+        async def __aenter__(self) -> Path:
+            self.__pwd = await Path.cwd()
+            await to_thread.run_sync(os.chdir, self.__cwd)
+            return self.__cwd
+
+        async def __aexit__(
+            self,
+            exc_type: type[BaseException] | None,
+            exc_value: BaseException | None,
+            traceback: TracebackType | None,
+        ) -> bool | None:
+            if self.__pwd is not None:
+                await to_thread.run_sync(os.chdir, self.__pwd)
+            return await super().__aexit__(exc_type, exc_value, traceback)
+
     @pytest.fixture
     def populated_tmpdir(self, tmp_path: pathlib.Path) -> pathlib.Path:
         tmp_path.joinpath("testfile").touch()
         tmp_path.joinpath("testfile2").touch()
+
         subdir = tmp_path / "subdir"
-        subdir.mkdir()
-        subdir.joinpath("dummyfile1.txt").touch()
-        subdir.joinpath("dummyfile2.txt").touch()
+        sibdir = tmp_path / "sibdir"
+
+        for subpath in (subdir, sibdir):
+            subpath.mkdir()
+            subpath.joinpath("dummyfile1.txt").touch()
+            subpath.joinpath("dummyfile2.txt").touch()
+
         return tmp_path
 
     async def test_properties(self) -> None:
@@ -297,19 +325,29 @@ async def test_glob(self, populated_tmpdir: pathlib.Path) -> None:
         all_paths = []
         async for path in Path(populated_tmpdir).glob("**/*.txt"):
             assert isinstance(path, Path)
-            all_paths.append(path.name)
+            all_paths.append(path.relative_to(populated_tmpdir))
 
         all_paths.sort()
-        assert all_paths == ["dummyfile1.txt", "dummyfile2.txt"]
+        assert all_paths == [
+            Path("sibdir") / "dummyfile1.txt",
+            Path("sibdir") / "dummyfile2.txt",
+            Path("subdir") / "dummyfile1.txt",
+            Path("subdir") / "dummyfile2.txt",
+        ]
 
     async def test_rglob(self, populated_tmpdir: pathlib.Path) -> None:
         all_paths = []
         async for path in Path(populated_tmpdir).rglob("*.txt"):
             assert isinstance(path, Path)
-            all_paths.append(path.name)
+            all_paths.append(path.relative_to(populated_tmpdir))
 
         all_paths.sort()
-        assert all_paths == ["dummyfile1.txt", "dummyfile2.txt"]
+        assert all_paths == [
+            Path("sibdir") / "dummyfile1.txt",
+            Path("sibdir") / "dummyfile2.txt",
+            Path("subdir") / "dummyfile1.txt",
+            Path("subdir") / "dummyfile2.txt",
+        ]
 
     async def test_iterdir(self, populated_tmpdir: pathlib.Path) -> None:
         all_paths = []
@@ -318,7 +356,7 @@ async def test_iterdir(self, populated_tmpdir: pathlib.Path) -> None:
             all_paths.append(path.name)
 
         all_paths.sort()
-        assert all_paths == ["subdir", "testfile", "testfile2"]
+        assert all_paths == ["sibdir", "subdir", "testfile", "testfile2"]
 
     def test_joinpath(self) -> None:
         path = Path("/foo").joinpath("bar")
@@ -421,10 +459,25 @@ async def test_read_text(self, tmp_path: pathlib.Path) -> None:
         path.write_text("some text åäö", encoding="utf-8")
         assert await Path(path).read_text(encoding="utf-8") == "some text åäö"
 
-    async def test_relative_to(self, tmp_path: pathlib.Path) -> None:
+    async def test_relative_to_subpath(self, tmp_path: pathlib.Path) -> None:
         path = tmp_path / "subdir"
         assert path.relative_to(tmp_path) == Path("subdir")
 
+    @pytest.mark.skipif(
+        sys.version_info < (3, 12),
+        reason="Path.relative_to(walk_up=<bool>) is only available on Python 3.12+",
+    )
+    async def test_relative_to_sibling(self, populated_tmpdir: pathlib.Path) -> None:
+        subdir = Path(populated_tmpdir / "subdir")
+        sibdir = Path(populated_tmpdir / "sibdir")
+
+        with pytest.raises(ValueError):
+            subdir.relative_to(sibdir, walk_up=False)
+
+        async with TestPath.ChangeDirectory(sibdir) as path:
+            relpath = subdir.relative_to(path, walk_up=True) / "dummyfile1.txt"
+            assert os.access(relpath, os.R_OK)
+
     async def test_rename(self, tmp_path: pathlib.Path) -> None:
         path = tmp_path / "somefile"
         path.touch()