test: handle incomplete descriptor directory views

docs/changelog/+partial-fd-view.bugfix.rst

@@ -0,0 +1 @@
+Fix descriptor-counting tests and inode-reuse fixtures when descriptor directories only list the standard streams.

tests/conftest.py

@@ -17,6 +17,7 @@
 import filelock._identity
 from filelock import BaseFileLock, MarkerSoftFileLock, SoftFileLease, SoftFileLock, Timeout
 from filelock._identity import host_name
+from tests.descriptor_helpers import descriptor_ids
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Generator, Iterator
@@ -191,9 +192,9 @@ def lstat_with_reuse(path: str | os.PathLike[str], *, dir_fd: int | None = None)
 
 
 def _held_open(node: os.stat_result) -> bool:  # pragma: win32 no cover
-    for entry in Path("/dev/fd").iterdir():
+    for descriptor_id in descriptor_ids():
         with suppress(OSError):  # the descriptor the listing itself used is closed by now
-            if os.path.samestat(os.fstat(int(entry.name)), node):
+            if os.path.samestat(os.fstat(descriptor_id), node):
                 return True
     return False
 

tests/descriptor_helpers.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+import os
+from contextlib import suppress
+from pathlib import Path
+
+
+def descriptor_ids() -> tuple[int, ...] | range:  # pragma: needs fd-directory
+    """Find descriptor candidates even when /dev/fd only lists the standard streams."""
+    with Path(os.devnull).open("rb") as probe:
+        for view in (Path("/dev/fd"), Path("/proc/self/fd")):
+            with suppress(OSError):
+                descriptors = tuple(int(entry.name) for entry in view.iterdir() if entry.name.isdecimal())
+                if probe.fileno() >= 3 and probe.fileno() in descriptors:
+                    return descriptors
+    return range(os.sysconf("SC_OPEN_MAX"))

tests/test_descriptor_views.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import TYPE_CHECKING, Final
+
+import pytest
+from capabilities import CAPABILITIES
+
+from tests.conftest import _held_open
+from tests.test_read_write_fork import _descriptors_on as read_write_descriptors
+from tests.test_strict_soft_failures import _descriptors_on as strict_soft_descriptors
+
+if TYPE_CHECKING:
+    from collections.abc import Callable, Iterator
+
+pytestmark = pytest.mark.skipif(not CAPABILITIES["fd-directory"], reason="requires a descriptor directory")
+
+
+@pytest.fixture(params=["complete", "partial", "missing", "proc-only"])
+def descriptor_view(
+    request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch
+) -> None:  # pragma: needs fd-directory
+    iterdir: Final = Path.iterdir
+    native: Final = Path("/dev/fd") if Path("/dev/fd").is_dir() else Path("/proc/self/fd")
+    limit: Final = max(int(entry.name) for entry in iterdir(native)) + 64
+
+    def entries(path: Path) -> Iterator[Path]:
+        if path == Path("/dev/fd"):
+            if request.param == "complete":
+                return iterdir(native)
+            if request.param in {"partial", "proc-only"}:
+                return iter(path / str(number) for number in range(3))
+            raise FileNotFoundError(path)
+        if path == Path("/proc/self/fd"):
+            if request.param == "proc-only":
+                return iterdir(native)
+            raise FileNotFoundError(path)
+        return iterdir(path)
+
+    monkeypatch.setattr(Path, "iterdir", entries)
+    monkeypatch.setattr(os, "sysconf", lambda _name: limit)
+
+
+@pytest.mark.usefixtures("descriptor_view")
+@pytest.mark.parametrize("count", [read_write_descriptors, strict_soft_descriptors])
+def test_descriptor_counts_follow_real_open_files(  # pragma: needs fd-directory
+    tmp_path: Path, count: Callable[[Path], int]
+) -> None:
+    path: Final = tmp_path / "held"
+    with path.open("w+b") as first:
+        second = os.dup(first.fileno())
+        try:
+            assert count(path) == 2
+        finally:
+            os.close(second)
+        assert count(path) == 1
+    assert count(path) == 0
+
+
+@pytest.mark.usefixtures("descriptor_view")
+def test_inode_reuse_waits_for_an_unlinked_files_last_descriptor(tmp_path: Path) -> None:  # pragma: needs fd-directory
+    path: Final = tmp_path / "unlinked"
+    with path.open("w+b") as held:
+        node = os.fstat(held.fileno())
+        path.unlink()
+        assert _held_open(node)
+    assert not _held_open(node)
+
+
+@pytest.mark.usefixtures("descriptor_view")
+def test_descriptor_view_preserves_unrelated_directory_entries(tmp_path: Path) -> None:  # pragma: needs fd-directory
+    child = tmp_path / "ordinary-file"
+    child.touch()
+    assert list(tmp_path.iterdir()) == [child]

tests/test_read_write_fork.py

@@ -22,6 +22,7 @@
     NEEDS_FORK,
     NEEDS_FORK1,
 )
+from tests.descriptor_helpers import descriptor_ids
 
 if TYPE_CHECKING:
     from _typeshed import StrOrBytesPath
@@ -535,15 +536,7 @@ def _dropped_instances_script() -> str:  # pragma: needs fd-directory
 def _descriptors_on(path: Path) -> int:  # pragma: needs fd-directory
     database: Final[os.stat_result] = path.stat()
     count = 0
-    descriptor_ids: Final = (
-        range(os.sysconf("SC_OPEN_MAX"))
-        if sys.platform == "netbsd"
-        else (
-            int(entry.name)
-            for entry in (Path("/dev/fd") if Path("/dev/fd").is_dir() else Path("/proc/self/fd")).iterdir()
-        )
-    )
-    for descriptor_id in descriptor_ids:
+    for descriptor_id in descriptor_ids():
         with contextlib.suppress(OSError):
             descriptor: Final[os.stat_result] = os.fstat(descriptor_id)
             count += (descriptor.st_dev, descriptor.st_ino) == (database.st_dev, database.st_ino)

tests/test_strict_soft_failures.py

@@ -29,6 +29,7 @@
     NEEDS_SYMLINK,
     NEEDS_UNLINK_OPEN_FILE,
 )
+from tests.descriptor_helpers import descriptor_ids
 
 # These cases inject a failure into a directory-fd cleanup step or the reaper race, both of which only run where the
 # protocol uses dir_fd descriptors. Without dir_fd those branches never execute and the injected fault never fires; the
@@ -1929,10 +1930,10 @@ def deny(path: _PathValue, *, dir_fd: int | None = None) -> None:
 def _descriptors_on(path: Path) -> int:  # pragma: needs fd-directory
     target: Final = path.stat()
     count = 0
-    for entry in (Path("/dev/fd") if Path("/dev/fd").is_dir() else Path("/proc/self/fd")).iterdir():
+    for descriptor_id in descriptor_ids():
         # The descriptor that listed the directory is already closed.
         with contextlib.suppress(OSError):
-            count += os.path.samestat(os.fstat(int(entry.name)), target)
+            count += os.path.samestat(os.fstat(descriptor_id), target)
     return count