✅ test: gate GraalPy gaps on probed capabilities

tasks/coverage_pragmas.py

@@ -14,23 +14,38 @@
 
 from __future__ import annotations
 
+import gc
 import os
 import signal
 import socket
+import sys
 import tempfile
+import weakref
+from asyncio import CancelledError
+from contextlib import asynccontextmanager, contextmanager
 from importlib.util import find_spec
 from pathlib import Path
-from typing import TYPE_CHECKING, Final, cast
+from typing import TYPE_CHECKING, Any, Final, cast
 
 from coverage import CoveragePlugin
 
 if TYPE_CHECKING:
-    from collections.abc import Iterable
+    from collections.abc import AsyncIterator, Generator, Iterable, Iterator
 
     from coverage.plugin_support import Plugins
     from coverage.types import TConfigurable
 
 
+class _Suspend:
+    """An awaitable that parks its coroutine once, so a probe can throw into a suspended frame without a loop."""
+
+    def __await__(self) -> Generator[None, Any, None]:
+        yield
+
+
+_AUDIT_PROBE_EVENT: Final[str] = "filelock.capability-probe"
+
+
 # coverage passes the config's plugin options; this plugin takes none.
 def coverage_init(reg: Plugins, options: dict[str, str]) -> None:  # ruff:ignore[unused-function-argument]
     reg.add_configurer(CapabilityPragmas())
@@ -102,6 +117,111 @@ def _honors_link_follow_symlinks() -> bool:
         return True
 
 
+def _finalizes_on_last_reference() -> bool:
+    # Only a refcounting collector runs __del__ the moment the last reference goes; a tracing one defers it.
+    finalized: list[bool] = []
+
+    class Probe:
+        def __del__(self) -> None:
+            finalized.append(True)
+
+    Probe()
+    return bool(finalized)
+
+
+def _finalizes_on_collection() -> bool:
+    # GraalPy queues finalizers on the host collector, so even a forced collection does not run __del__.
+    finalized: list[bool] = []
+
+    class Probe:
+        def __del__(self) -> None:
+            finalized.append(True)
+
+    Probe()
+    gc.collect()
+    return bool(finalized)
+
+
+def _collects_classes() -> bool:
+    # A dynamically built lock subclass must not outlive its last reference, or the registries keyed on it leak.
+    def build() -> type:
+        class Probe:
+            pass
+
+        return Probe
+
+    reference = weakref.ref(build())
+    gc.collect()
+    return reference() is None
+
+
+def _preserves_context_thrown_into_a_generator() -> bool:
+    # GraalPy resets __context__ when contextlib throws into the suspended generator, losing the chained cause.
+    @contextmanager
+    def probe() -> Iterator[None]:
+        yield
+
+    error = KeyError("thrown")
+    error.__context__ = ValueError("context")
+    try:
+        with probe():
+            raise error
+    except KeyError as caught:
+        return caught.__context__ is not None
+    return False  # pragma: no cover  # the raise above always propagates
+
+
+def _propagates_a_cancellation_thrown_into_a_coroutine() -> bool:
+    # Driven by hand so the probe needs no event loop: GraalPy answers athrow with RuntimeError instead of the
+    # CancelledError, so every cancellation crossing an async context manager surfaces as the wrong exception.
+    @asynccontextmanager
+    async def gate() -> AsyncIterator[None]:
+        await _Suspend()
+        yield
+
+    async def body() -> None:
+        async with gate():
+            pass
+
+    coroutine = body()
+    coroutine.send(None)
+    try:
+        coroutine.throw(CancelledError())
+    except CancelledError:
+        return True
+    except BaseException:  # ruff:ignore[blind-except]  # whatever else a runtime substitutes counts as the deviation
+        return False
+    return False  # pragma: no cover  # throwing into the suspended coroutine always raises
+
+
+def _delivers_audit_events() -> bool:
+    # GraalPy accepts a hook and never calls it. The hook is a permanent no-op; tests install their own anyway.
+    delivered: list[bool] = []
+    sys.addaudithook(lambda event, _args: delivered.append(True) if event == _AUDIT_PROBE_EVENT else None)
+    sys.audit(_AUDIT_PROBE_EVENT)
+    return bool(delivered)
+
+
+def _refuses_to_open_a_symlink() -> bool:
+    # GraalPy accepts O_NOFOLLOW then follows the link anyway, so ask for the refusal rather than the constant.
+    if not hasattr(os, "O_NOFOLLOW"):
+        return False
+    if not _supports_symlink():
+        # Nothing to point the probe at, so keep the constant's answer rather than reporting a gap we cannot see.
+        return True
+    with tempfile.TemporaryDirectory() as directory:
+        target = Path(directory, "target")
+        target.touch()
+        link = Path(directory, "link")
+        link.symlink_to(target)
+        try:
+            descriptor = os.open(link, os.O_RDONLY | os.O_NOFOLLOW)
+        except OSError:
+            return True
+        os.close(descriptor)
+        return False
+
+
 def _enforces_file_mode() -> bool:
     # Without POSIX permission bits a chmod does not read back.
     with tempfile.TemporaryDirectory() as directory:
@@ -117,17 +237,25 @@ def _enforces_file_mode() -> bool:
 CAPABILITIES: Final[dict[str, bool]] = {
     "fork": hasattr(os, "fork") and hasattr(os, "register_at_fork"),
     "dir-fd": os.open in os.supports_dir_fd,
+    # Narrower than "dir-fd": GraalPy takes os.open relative to a directory descriptor but not os.link.
+    "link-dir-fd": hasattr(os, "link") and os.link in os.supports_dir_fd,
     "hard-link": hasattr(os, "link"),
     "symlink": _supports_symlink(),
     "fcntl": find_spec("fcntl") is not None,
     "unlink-open-file": _supports_unlinking_an_open_file(),
     "posix-signals": hasattr(signal, "SIGKILL"),
     "file-mode": _enforces_file_mode(),
+    "prompt-finalization": _finalizes_on_last_reference(),
+    "collected-finalization": _finalizes_on_collection(),
+    "class-collection": _collects_classes(),
+    "generator-exception-context": _preserves_context_thrown_into_a_generator(),
+    "coroutine-cancellation": _propagates_a_cancellation_thrown_into_a_coroutine(),
+    "audit-events": _delivers_audit_events(),
     "fd-directory": any(Path(view).is_dir() for view in ("/dev/fd", "/proc/self/fd")),
     "fifo": hasattr(os, "mkfifo"),
     "af-unix": hasattr(socket, "AF_UNIX"),
     # Distinct from "symlink": a runtime can create them yet still not refuse to follow one.
-    "o-nofollow": hasattr(os, "O_NOFOLLOW"),
+    "o-nofollow": _refuses_to_open_a_symlink(),
     "utime-nofollow": os.utime in os.supports_follow_symlinks,
     "utime-fd": os.utime in os.supports_fd,
     "sqlite3": find_spec("sqlite3") is not None,

tests/capability_marks.py

@@ -0,0 +1,59 @@
+"""Marks for runtime capabilities several test modules gate on.
+
+Capabilities used by a single module stay private to it, as ``_NEEDS_SYMLINK`` and friends do. These span modules, so
+they live here and read the same ``CAPABILITIES`` probes the coverage pragmas use.
+"""
+
+from __future__ import annotations
+
+import pytest
+from coverage_pragmas import CAPABILITIES
+
+#: A dropped reference releases what its __del__ releases. Only a refcounting collector does this.
+NEEDS_PROMPT_FINALIZATION = pytest.mark.skipif(
+    not CAPABILITIES["prompt-finalization"],
+    reason="a dropped reference does not run __del__ on a deferred collector",
+)
+
+#: gc.collect() runs pending __del__ methods. GraalPy hands them to the host collector and never gets them back.
+NEEDS_COLLECTED_FINALIZATION = pytest.mark.skipif(
+    not CAPABILITIES["collected-finalization"],
+    reason="gc.collect() does not run __del__ on this runtime",
+)
+
+#: gc.collect() reclaims a dynamically built class once its last reference goes.
+NEEDS_CLASS_COLLECTION = pytest.mark.skipif(
+    not CAPABILITIES["class-collection"],
+    reason="gc.collect() does not reclaim classes on this runtime",
+)
+
+#: An exception thrown into a suspended generator or coroutine keeps its __context__.
+NEEDS_GENERATOR_EXCEPTION_CONTEXT = pytest.mark.skipif(
+    not CAPABILITIES["generator-exception-context"],
+    reason="this runtime clears __context__ when an exception is thrown into a suspended frame",
+)
+
+#: sys.audit delivers to hooks installed with sys.addaudithook. GraalPy accepts the hook and never calls it.
+NEEDS_AUDIT_EVENTS = pytest.mark.skipif(
+    not CAPABILITIES["audit-events"],
+    reason="this runtime never delivers audit events to an installed hook",
+)
+
+#: A cancellation crossing an async context manager surfaces as CancelledError rather than the interpreter's own
+#: bookkeeping error. GraalPy's contextlib raises ``RuntimeError: generator didn't stop after athrow()`` from
+#: ``_GeneratorContextManagerBase.__aexit__`` instead, so the lock's cancellation contract cannot be observed there.
+#: Not strict: the deviation depends on where the cancellation lands, so some of these tests still pass.
+XFAIL_WITHOUT_COROUTINE_CANCELLATION = pytest.mark.xfail(
+    not CAPABILITIES["coroutine-cancellation"],
+    reason="GraalPy's contextlib answers athrow() with RuntimeError instead of propagating the CancelledError",
+    strict=False,
+)
+
+__all__ = [
+    "NEEDS_AUDIT_EVENTS",
+    "NEEDS_CLASS_COLLECTION",
+    "NEEDS_COLLECTED_FINALIZATION",
+    "NEEDS_GENERATOR_EXCEPTION_CONTEXT",
+    "NEEDS_PROMPT_FINALIZATION",
+    "XFAIL_WITHOUT_COROUTINE_CANCELLATION",
+]

tests/soft_rw/test_soft_rw_sync.py

@@ -731,8 +731,8 @@ def test_stale_malformed_marker_is_evicted(lock_file: str, content: bytes) -> No
 
 
 def test_fifo_write_marker_does_not_block(lock_file: str) -> None:  # pragma: needs fifo
-    if sys.platform == "win32":  # pragma: win32 cover
-        pytest.skip("os.mkfifo is unix-only")  # also narrows sys.platform so ty resolves os.mkfifo below
+    if sys.platform == "win32" or not CAPABILITIES["fifo"]:  # pragma: win32 cover
+        pytest.skip("os.mkfifo is unavailable")  # the platform arm also narrows so ty resolves os.mkfifo below
     marker = f"{lock_file}.write"
     os.mkfifo(marker)
     past = time.time() - 1000
@@ -747,8 +747,8 @@ def test_fifo_write_marker_does_not_block(lock_file: str) -> None:  # pragma: ne
 
 
 def test_fifo_write_marker_with_writer_is_evicted(lock_file: str) -> None:  # pragma: needs fifo
-    if sys.platform == "win32":  # pragma: win32 cover
-        pytest.skip("os.mkfifo is unix-only")  # also narrows sys.platform so ty resolves os.mkfifo below
+    if sys.platform == "win32" or not CAPABILITIES["fifo"]:  # pragma: win32 cover
+        pytest.skip("os.mkfifo is unavailable")  # the platform arm also narrows so ty resolves os.mkfifo below
     marker = f"{lock_file}.write"
     os.mkfifo(marker)
     past = time.time() - 1000
@@ -797,7 +797,7 @@ def test_touch_does_not_follow_symlink(lock_file: str, tmp_path: Path) -> None:
 
     util_mod.touch(str(marker))
 
-    assert victim.stat().st_mtime == past
+    assert victim.stat().st_mtime == pytest.approx(past)  # a timestamp round-trip need not be bit-exact
     assert victim.read_text() == "do-not-touch"
 
 
@@ -827,7 +827,7 @@ def swap_after_open(name: str, *, dir_fd: int | None = None) -> int | None:
 
         monkeypatch.setattr(sync_mod, "_open_marker", swap_after_open)
         assert lock._refresh_marker() is True
-        assert victim.stat().st_mtime == past
+        assert victim.stat().st_mtime == pytest.approx(past)  # a timestamp round-trip need not be bit-exact
         assert victim.read_text() == "do-not-touch"
     finally:
         lock.release(force=True)

tests/test_async_filelock_acquire_cancellation.py

@@ -11,6 +11,7 @@
 
 import pytest
 from async_filelock_cancellation_helpers import assert_cancellation_message, assert_file_lock_state
+from capability_marks import XFAIL_WITHOUT_COROUTINE_CANCELLATION
 
 from filelock import (
     AsyncFileLock,
@@ -56,6 +57,7 @@ async def _release(self) -> None:  # ty: ignore[invalid-method-override]
 
 @pytest.mark.parametrize("policy", [pytest.param("chain", id="chain"), pytest.param("group", id="group")])
 @pytest.mark.asyncio
+@XFAIL_WITHOUT_COROUTINE_CANCELLATION
 async def test_backend_cancellation_rollback_failure_follows_policy(tmp_path: Path, policy: ContextErrorPolicy) -> None:
     backend_cancellation = asyncio.CancelledError("backend canceled")
     rollback_error = OSError(EIO, "rollback failed")
@@ -185,6 +187,7 @@ def block_hook(_fd: int) -> None:
 
 @_NEEDS_FCNTL
 @pytest.mark.asyncio
+@XFAIL_WITHOUT_COROUTINE_CANCELLATION
 async def test_cancelled_queued_acquire_does_not_claim_transition(tmp_path: Path) -> None:  # pragma: needs fcntl
     hook_started = asyncio.Event()
     finish_hook = threading.Event()
@@ -251,6 +254,7 @@ def block_rollback(_fd: int, _operation: int) -> None:
 @_NEEDS_FCNTL
 @pytest.mark.parametrize("policy", [pytest.param("chain", id="chain"), pytest.param("group", id="group")])
 @pytest.mark.asyncio
+@XFAIL_WITHOUT_COROUTINE_CANCELLATION
 async def test_acquire_cancellation_surfaces_attempt_error_after_rollback(  # pragma: needs fcntl
     tmp_path: Path, policy: ContextErrorPolicy
 ) -> None:
@@ -299,6 +303,7 @@ def fail_hook(_fd: int) -> None:
     [pytest.param("unrelated", True, id="distinct"), pytest.param("attempt", False, id="equivalent")],
 )
 @pytest.mark.asyncio
+@XFAIL_WITHOUT_COROUTINE_CANCELLATION
 async def test_acquire_cancellation_group_reconciles_attempt_context(  # pragma: needs fcntl
     tmp_path: Path, context_message: str, *, preserved: bool
 ) -> None:
@@ -347,6 +352,7 @@ def fail_hook(_fd: int) -> None:
 @_NEEDS_FCNTL
 @pytest.mark.parametrize("policy", [pytest.param("chain", id="chain"), pytest.param("group", id="group")])
 @pytest.mark.asyncio
+@XFAIL_WITHOUT_COROUTINE_CANCELLATION
 async def test_acquire_cancellation_surfaces_rollback_error(  # pragma: needs fcntl
     tmp_path: Path,
     mocker: MockerFixture,
@@ -395,6 +401,7 @@ def block_hook(_fd: int) -> None:
 @_NEEDS_FCNTL
 @pytest.mark.parametrize("policy", [pytest.param("chain", id="chain"), pytest.param("group", id="group")])
 @pytest.mark.asyncio
+@XFAIL_WITHOUT_COROUTINE_CANCELLATION
 async def test_acquire_cancellation_surfaces_attempt_and_rollback_errors(  # pragma: needs fcntl
     tmp_path: Path, mocker: MockerFixture, policy: ContextErrorPolicy
 ) -> None:

tests/test_async_filelock_release_cancellation.py

@@ -15,6 +15,7 @@
     get_fcntl,
     start_file_lock_holder,
 )
+from capability_marks import XFAIL_WITHOUT_COROUTINE_CANCELLATION
 
 from filelock import AsyncFileLock, BaseAsyncFileLock, ContextErrorPolicy
 
@@ -59,6 +60,7 @@ def block_unlock(_fd: int, _operation: int) -> None:
 
 @_NEEDS_FCNTL
 @pytest.mark.asyncio  # pragma: needs fcntl
+@XFAIL_WITHOUT_COROUTINE_CANCELLATION
 async def test_acquire_proceeds_after_queued_release_is_canceled(tmp_path: Path, mocker: MockerFixture) -> None:
     lock = AsyncFileLock(tmp_path / "a")
     await lock.acquire()
@@ -158,6 +160,7 @@ def observe_first_poll() -> bool:
 @_NEEDS_FCNTL
 @pytest.mark.parametrize("policy", [pytest.param("chain", id="chain"), pytest.param("group", id="group")])
 @pytest.mark.asyncio
+@XFAIL_WITHOUT_COROUTINE_CANCELLATION
 async def test_release_cancellation_surfaces_backend_error(  # pragma: needs fcntl
     tmp_path: Path, mocker: MockerFixture, caplog: pytest.LogCaptureFixture, policy: ContextErrorPolicy
 ) -> None:

tests/test_async_filelock_rollback_agnostic.py

@@ -6,6 +6,7 @@
 from typing import TYPE_CHECKING
 
 import pytest
+from capability_marks import XFAIL_WITHOUT_COROUTINE_CANCELLATION
 
 from filelock import AsyncFileLock, BaseAsyncFileLock
 
@@ -50,6 +51,7 @@ async def _release(self) -> None:  # ty: ignore[invalid-method-override]
 
 
 @pytest.mark.asyncio
+@XFAIL_WITHOUT_COROUTINE_CANCELLATION
 async def test_release_cancellation_surfaces_backend_error(tmp_path: Path) -> None:
     backend_error = OSError(EIO, "backend release failed")
     release_started = asyncio.Event()
@@ -86,6 +88,7 @@ async def _release(self) -> None:  # ty: ignore[invalid-method-override]
 
 
 @pytest.mark.asyncio
+@XFAIL_WITHOUT_COROUTINE_CANCELLATION
 async def test_acquire_cancellation_rollback_failure_surfaces_backend_error(tmp_path: Path) -> None:
     rollback_error = OSError(EIO, "rollback release failed")
     acquire_started = asyncio.Event()

tests/test_async_filelock_runner_shutdown.py

@@ -10,6 +10,7 @@
 
 import pytest
 from async_filelock_cancellation_helpers import assert_file_lock_state, get_fcntl
+from capability_marks import XFAIL_WITHOUT_COROUTINE_CANCELLATION
 
 from filelock import AsyncAcquireReturnProxy, AsyncFileLock, ContextErrorPolicy
 
@@ -78,6 +79,7 @@ def block_hook(_fd: int) -> None:
 
 @_NEEDS_FCNTL
 @pytest.mark.parametrize("policy", [pytest.param("chain", id="chain"), pytest.param("group", id="group")])
+@XFAIL_WITHOUT_COROUTINE_CANCELLATION
 def test_runner_shutdown_preserves_body_cancellation_and_release_errors(  # pragma: needs fcntl
     tmp_path: Path, mocker: MockerFixture, policy: ContextErrorPolicy
 ) -> None: