Repository: agronholm/anyio

Issue:
Allow passing kwargs to anyio.open_process and anyio.run_process
### Things to check first

- [X] I have searched the existing issues and didn't find my feature already requested there


### Feature description

Accept `**kwargs` in `anyio.run_process` and `anyio.open_process` and pass those to underlying `asyncio.create_subprocess_exec`, `asyncio.create_subprocess_shell` or `trio._subprocess.open_process`

### Use case

`subprocess.Popen` allows to pass a number of kwargs to manage process behavior (most of which are OS-dependent)

```
def __init__(self, args, bufsize=-1, executable=None,
                 stdin=None, stdout=None, stderr=None,
                 preexec_fn=None, close_fds=True,
                 shell=False, cwd=None, env=None, universal_newlines=None,
                 startupinfo=None, creationflags=0,
                 restore_signals=True, start_new_session=False,
                 pass_fds=(), *, user=None, group=None, extra_groups=None,
                 encoding=None, errors=None, text=None, umask=-1, pipesize=-1,
                 process_group=None):
```

Currently it seems not possible to pass most of these to anyio's wrappers `anyio.open_process` and `anyio.run_process`, but it would be beneficial to have.

Discussion:
dolamroth: Tests 


1.py
```
import sys
sys.setrecursionlimit(1000)

def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)

print(fib(200))
sys.exit(0)
```

test.py
```
import anyio
import subprocess
import resource
import sys

def limit_virtual_memory():
    resource.setrlimit(resource.RLIMIT_CPU, (1, 1))

async def main():
    try:
        proc = await anyio.run_process([sys.executable, "1.py"], preexec_fn=limit_virtual_memory, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        print(proc.stdout)
    except BaseException as exc:
        print(exc)


if __name__ == "__main__":
    anyio.run(main)
```

test_sync.py
```
import subprocess
import resource
import sys

def limit_virtual_memory():
    resource.setrlimit(resource.RLIMIT_CPU, (1, 1))

def main():
    try:
        proc = subprocess.run([sys.executable, "1.py"], preexec_fn=limit_virtual_memory, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        print(proc.stdout)
    except BaseException as exc:
        print(exc)


if __name__ == "__main__":
    main()
```

test.py prints:
```
Command '['/home/user/venv/bin/python', '1.py']' died with <Signals.SIGKILL: 9>.
```

test_sync.py prints:
```
b''
```
dolamroth: Test 2: switch back to limiting RAM

```
import anyio
import subprocess
import resource
import sys

MAX_VIRTUAL_MEMORY = 10 * 1024 * 1024


def limit_virtual_memory():
    resource.setrlimit(resource.RLIMIT_AS, (MAX_VIRTUAL_MEMORY, MAX_VIRTUAL_MEMORY))

async def main():
    try:
        proc = await anyio.run_process([sys.executable, "1.py"], preexec_fn=limit_virtual_memory, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        print(proc.stdout)
    except BaseException as exc:
        print(exc)

if __name__ == "__main__":
    anyio.run(main)
```

If I set `MAX_VIRTUAL_MEMORY = 10 * 1024`, test.py outputs:
```
Command '['/home/user/venv/bin/python', '1.py']' died with <Signals.SIGSEGV: 11>.
```

but if I set it to a larger number, it for some reason outputs:
```
Command '['/home/user/venv/bin/python', '1.py']' returned non-zero exit status 127.
```

This doesn't happen on the same setup in `test_sync.py`
dolamroth: Test of trio

```
import trio
import subprocess
import resource
import sys
from functools import partial

MAX_VIRTUAL_MEMORY = 10 * 1024 * 1024

def limit_virtual_memory():
    resource.setrlimit(resource.RLIMIT_AS, (MAX_VIRTUAL_MEMORY, MAX_VIRTUAL_MEMORY))

async def main():
    try:
        async with trio.open_nursery() as nursery:
            start_proc = partial(trio.run_process, preexec_fn=limit_virtual_memory, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            proc = await nursery.start(start_proc, [sys.executable, "1.py"])
        print(proc.stdout)
    except BaseException as exc:
        print(exc.exceptions)

if __name__ == "__main__":
    trio.run(main)
```

Outputs
```
(CalledProcessError(127, ['/home/.../venv/bin/python', '1.py']),)
```
dolamroth: test_asyncio.py

```
import asyncio
import subprocess
import resource
import sys
from functools import partial

MAX_VIRTUAL_MEMORY = 10 * 1024 * 1024

def limit_virtual_memory():
    resource.setrlimit(resource.RLIMIT_AS, (MAX_VIRTUAL_MEMORY, MAX_VIRTUAL_MEMORY))

async def main():
    try:
        proc = await asyncio.create_subprocess_exec(*[sys.executable, "1.py"], preexec_fn=limit_virtual_memory, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        print(proc.stdout)
    except BaseException as exc:
        print(exc)

if __name__ == "__main__":
    asyncio.run(main())
```

Outputs
```
<StreamReader eof transport=<_UnixReadPipeTransport closed fd=6 closed>>
```
agronholm: > Comparing to the full list of `Popen` parameters:
> 
> ```python
> # Supported:
> args,
> stdin,
> stdout,
> stderr,
> cwd,
> env,
> startupinfo,
> creationflags,
> start_new_session,
> pass_fds,
> user,
> group,
> extra_groups,
> umask,
> 
> # Not supported:
> preexec_fn,  # Barely supported by Python (unsafe with threads)
> close_fds,  # Unsafe (just set it to True and have users use `pass_fds` and `startupinfo.lpAttributeList["handle_list"]` instead)
> # Not supported by asyncio:
> bufsize,
> universal_newlines,
> shell,
> text,
> encoding,
> errors,
> # _Possible_ remaining contenders to support in AnyIO (AFAIK nobody has requested any of
> # these yet):
> process_group,  # Similar to `start_new_session` in that it can replace `preexec_fn` use cases.
> executable,
> restore_signals,
> pipesize,  # Not super useful since if using pipes you should have tasks always reading from the std{out,err} streams.
> ```
> 
> Personally I do not have a use-case for any of those four at the bottom right now, but I wanted to ask why you excluded them.

Many are outright dangerous, or problematic from the type annotations PoV. I tried to include a larger number of these but gave up on the annotations front.

Patch:
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index 73463d9..2eb72bc 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -10,6 +10,12 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 - Added support for the ``from_uri()``, ``full_match()``, ``parser`` methods/properties
   in ``anyio.Path``, newly added in Python 3.13
   (`#737 <https://github.com/agronholm/anyio/issues/737>`_)
+- Added support for more keyword arguments for ``run_process()`` and ``open_process()``:
+  ``startupinfo``, ``creationflags``, ``pass_fds``, ``user``, ``group``,
+  ``extra_groups`` and ``umask``
+  (`#742 <https://github.com/agronholm/anyio/issues/742>`_)
+- Improved the type annotations and support for ``PathLike`` in ``run_process()`` and
+  ``open_process()`` to allow for path-like arguments, just like ``subprocess.Popen``
 - Changed the ``ResourceWarning`` from an unclosed memory object stream to include its
   address for easier identification
 - Changed ``start_blocking_portal()`` to always use daemonic threads, to accommodate the
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index b531ad4..b67f8e2 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -4,6 +4,7 @@ import array
 import asyncio
 import concurrent.futures
 import math
+import os
 import socket
 import sys
 import threading
@@ -47,7 +48,6 @@ from typing import (
     Collection,
     ContextManager,
     Coroutine,
-    Mapping,
     Optional,
     Sequence,
     Tuple,
@@ -81,6 +81,7 @@ from ..abc import (
     UDPPacketType,
     UNIXDatagramPacketType,
 )
+from ..abc._eventloop import StrOrBytesPath
 from ..lowlevel import RunVar
 from ..streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 
@@ -2239,26 +2240,24 @@ class AsyncIOBackend(AsyncBackend):
     @classmethod
     async def open_process(
         cls,
-        command: str | bytes | Sequence[str | bytes],
+        command: StrOrBytesPath | Sequence[StrOrBytesPath],
         *,
-        shell: bool,
         stdin: int | IO[Any] | None,
         stdout: int | IO[Any] | None,
         stderr: int | IO[Any] | None,
-        cwd: str | bytes | PathLike | None = None,
-        env: Mapping[str, str] | None = None,
-        start_new_session: bool = False,
+        **kwargs: Any,
     ) -> Process:
         await cls.checkpoint()
-        if shell:
+        if isinstance(command, PathLike):
+            command = os.fspath(command)
+
+        if isinstance(command, (str, bytes)):
             process = await asyncio.create_subprocess_shell(
-                cast("str | bytes", command),
+                command,
                 stdin=stdin,
                 stdout=stdout,
                 stderr=stderr,
-                cwd=cwd,
-                env=env,
-                start_new_session=start_new_session,
+                **kwargs,
             )
         else:
             process = await asyncio.create_subprocess_exec(
@@ -2266,9 +2265,7 @@ class AsyncIOBackend(AsyncBackend):
                 stdin=stdin,
                 stdout=stdout,
                 stderr=stderr,
-                cwd=cwd,
-                env=env,
-                start_new_session=start_new_session,
+                **kwargs,
             )
 
         stdin_stream = StreamWriterWrapper(process.stdin) if process.stdin else None
diff --git a/src/anyio/_backends/_trio.py b/src/anyio/_backends/_trio.py
index cf6f3db..6120500 100644
--- a/src/anyio/_backends/_trio.py
+++ b/src/anyio/_backends/_trio.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import array
 import math
+import os
 import socket
 import sys
 import types
@@ -25,7 +26,6 @@ from typing import (
     ContextManager,
     Coroutine,
     Generic,
-    Mapping,
     NoReturn,
     Sequence,
     TypeVar,
@@ -60,7 +60,7 @@ from .._core._synchronization import Event as BaseEvent
 from .._core._synchronization import ResourceGuard
 from .._core._tasks import CancelScope as BaseCancelScope
 from ..abc import IPSockAddrType, UDPPacketType, UNIXDatagramPacketType
-from ..abc._eventloop import AsyncBackend
+from ..abc._eventloop import AsyncBackend, StrOrBytesPath
 from ..streams.memory import MemoryObjectSendStream
 
 if sys.version_info >= (3, 10):
@@ -967,26 +967,39 @@ class TrioBackend(AsyncBackend):
     @classmethod
     async def open_process(
         cls,
-        command: str | bytes | Sequence[str | bytes],
+        command: StrOrBytesPath | Sequence[StrOrBytesPath],
         *,
-        shell: bool,
         stdin: int | IO[Any] | None,
         stdout: int | IO[Any] | None,
         stderr: int | IO[Any] | None,
-        cwd: str | bytes | PathLike | None = None,
-        env: Mapping[str, str] | None = None,
-        start_new_session: bool = False,
+        **kwargs: Any,
     ) -> Process:
-        process = await trio.lowlevel.open_process(  # type: ignore[misc]
-            command,  # type: ignore[arg-type]
-            stdin=stdin,
-            stdout=stdout,
-            stderr=stderr,
-            shell=shell,
-            cwd=cwd,
-            env=env,
-            start_new_session=start_new_session,
-        )
+        def convert_item(item: StrOrBytesPath) -> str:
+            str_or_bytes = os.fspath(item)
+            if isinstance(str_or_bytes, str):
+                return str_or_bytes
+            else:
+                return os.fsdecode(str_or_bytes)
+
+        if isinstance(command, (str, bytes, PathLike)):
+            process = await trio.lowlevel.open_process(
+                convert_item(command),
+                stdin=stdin,
+                stdout=stdout,
+                stderr=stderr,
+                shell=True,
+                **kwargs,
+            )
+        else:
+            process = await trio.lowlevel.open_process(
+                [convert_item(item) for item in command],
+                stdin=stdin,
+                stdout=stdout,
+                stderr=stderr,
+                shell=False,
+                **kwargs,
+            )
+
         stdin_stream = SendStreamWrapper(process.stdin) if process.stdin else None
         stdout_stream = ReceiveStreamWrapper(process.stdout) if process.stdout else None
         stderr_stream = ReceiveStreamWrapper(process.stderr) if process.stderr else None
diff --git a/src/anyio/_core/_subprocesses.py b/src/anyio/_core/_subprocesses.py
index 5d5d7b7..1ac2d54 100644
--- a/src/anyio/_core/_subprocesses.py
+++ b/src/anyio/_core/_subprocesses.py
@@ -1,26 +1,41 @@
 from __future__ import annotations
 
-from collections.abc import AsyncIterable, Mapping, Sequence
+import sys
+from collections.abc import AsyncIterable, Iterable, Mapping, Sequence
 from io import BytesIO
 from os import PathLike
 from subprocess import DEVNULL, PIPE, CalledProcessError, CompletedProcess
-from typing import IO, Any, cast
+from typing import IO, Any, Union, cast
 
 from ..abc import Process
 from ._eventloop import get_async_backend
 from ._tasks import create_task_group
 
+if sys.version_info >= (3, 10):
+    from typing import TypeAlias
+else:
+    from typing_extensions import TypeAlias
+
+StrOrBytesPath: TypeAlias = Union[str, bytes, "PathLike[str]", "PathLike[bytes]"]
+
 
 async def run_process(
-    command: str | bytes | Sequence[str | bytes],
+    command: StrOrBytesPath | Sequence[StrOrBytesPath],
     *,
     input: bytes | None = None,
     stdout: int | IO[Any] | None = PIPE,
     stderr: int | IO[Any] | None = PIPE,
     check: bool = True,
-    cwd: str | bytes | PathLike[str] | None = None,
+    cwd: StrOrBytesPath | None = None,
     env: Mapping[str, str] | None = None,
+    startupinfo: Any = None,
+    creationflags: int = 0,
     start_new_session: bool = False,
+    pass_fds: Sequence[int] = (),
+    user: str | int | None = None,
+    group: str | int | None = None,
+    extra_groups: Iterable[str | int] | None = None,
+    umask: int = -1,
 ) -> CompletedProcess[bytes]:
     """
     Run an external command in a subprocess and wait until it completes.
@@ -40,8 +55,20 @@ async def run_process(
         command
     :param env: if not ``None``, this mapping replaces the inherited environment
         variables from the parent process
+    :param startupinfo: an instance of :class:`subprocess.STARTUPINFO` that can be used
+        to specify process startup parameters (Windows only)
+    :param creationflags: flags that can be used to control the creation of the
+        subprocess (see :class:`subprocess.Popen` for the specifics)
     :param start_new_session: if ``true`` the setsid() system call will be made in the
         child process prior to the execution of the subprocess. (POSIX only)
+    :param pass_fds: sequence of file descriptors to keep open between the parent and
+        child processes. (POSIX only)
+    :param user: effective user to run the process as (Python >= 3.9, POSIX only)
+    :param group: effective group to run the process as (Python >= 3.9, POSIX only)
+    :param extra_groups: supplementary groups to set in the subprocess (Python >= 3.9,
+        POSIX only)
+    :param umask: if not negative, this umask is applied in the child process before
+        running the given command (Python >= 3.9, POSIX only)
     :return: an object representing the completed process
     :raises ~subprocess.CalledProcessError: if ``check`` is ``True`` and the process
         exits with a nonzero return code
@@ -62,7 +89,14 @@ async def run_process(
         stderr=stderr,
         cwd=cwd,
         env=env,
+        startupinfo=startupinfo,
+        creationflags=creationflags,
         start_new_session=start_new_session,
+        pass_fds=pass_fds,
+        user=user,
+        group=group,
+        extra_groups=extra_groups,
+        umask=umask,
     ) as process:
         stream_contents: list[bytes | None] = [None, None]
         async with create_task_group() as tg:
@@ -86,14 +120,21 @@ async def run_process(
 
 
 async def open_process(
-    command: str | bytes | Sequence[str | bytes],
+    command: StrOrBytesPath | Sequence[StrOrBytesPath],
     *,
     stdin: int | IO[Any] | None = PIPE,
     stdout: int | IO[Any] | None = PIPE,
     stderr: int | IO[Any] | None = PIPE,
-    cwd: str | bytes | PathLike[str] | None = None,
+    cwd: StrOrBytesPath | None = None,
     env: Mapping[str, str] | None = None,
+    startupinfo: Any = None,
+    creationflags: int = 0,
     start_new_session: bool = False,
+    pass_fds: Sequence[int] = (),
+    user: str | int | None = None,
+    group: str | int | None = None,
+    extra_groups: Iterable[str | int] | None = None,
+    umask: int = -1,
 ) -> Process:
     """
     Start an external command in a subprocess.
@@ -111,30 +152,58 @@ async def open_process(
     :param cwd: If not ``None``, the working directory is changed before executing
     :param env: If env is not ``None``, it must be a mapping that defines the
         environment variables for the new process
+    :param creationflags: flags that can be used to control the creation of the
+        subprocess (see :class:`subprocess.Popen` for the specifics)
+    :param startupinfo: an instance of :class:`subprocess.STARTUPINFO` that can be used
+        to specify process startup parameters (Windows only)
     :param start_new_session: if ``true`` the setsid() system call will be made in the
         child process prior to the execution of the subprocess. (POSIX only)
+    :param pass_fds: sequence of file descriptors to keep open between the parent and
+        child processes. (POSIX only)
+    :param user: effective user to run the process as (Python >= 3.9; POSIX only)
+    :param group: effective group to run the process as (Python >= 3.9; POSIX only)
+    :param extra_groups: supplementary groups to set in the subprocess (Python >= 3.9;
+        POSIX only)
+    :param umask: if not negative, this umask is applied in the child process before
+        running the given command (Python >= 3.9; POSIX only)
     :return: an asynchronous process object
 
     """
-    if isinstance(command, (str, bytes)):
-        return await get_async_backend().open_process(
-            command,
-            shell=True,
-            stdin=stdin,
-            stdout=stdout,
-            stderr=stderr,
-            cwd=cwd,
-            env=env,
-            start_new_session=start_new_session,
-        )
-    else:
-        return await get_async_backend().open_process(
-            command,
-            shell=False,
-            stdin=stdin,
-            stdout=stdout,
-            stderr=stderr,
-            cwd=cwd,
-            env=env,
-            start_new_session=start_new_session,
-        )
+    kwargs: dict[str, Any] = {}
+    if user is not None:
+        if sys.version_info < (3, 9):
+            raise TypeError("the 'user' argument requires Python 3.9 or later")
+
+        kwargs["user"] = user
+
+    if group is not None:
+        if sys.version_info < (3, 9):
+            raise TypeError("the 'group' argument requires Python 3.9 or later")
+
+        kwargs["group"] = group
+
+    if extra_groups is not None:
+        if sys.version_info < (3, 9):
+            raise TypeError("the 'extra_groups' argument requires Python 3.9 or later")
+
+        kwargs["extra_groups"] = group
+
+    if umask >= 0:
+        if sys.version_info < (3, 9):
+            raise TypeError("the 'umask' argument requires Python 3.9 or later")
+
+        kwargs["umask"] = umask
+
+    return await get_async_backend().open_process(
+        command,
+        stdin=stdin,
+        stdout=stdout,
+        stderr=stderr,
+        cwd=cwd,
+        env=env,
+        startupinfo=startupinfo,
+        creationflags=creationflags,
+        start_new_session=start_new_session,
+        pass_fds=pass_fds,
+        **kwargs,
+    )
diff --git a/src/anyio/abc/_eventloop.py b/src/anyio/abc/_eventloop.py
index a50afef..258d2e1 100644
--- a/src/anyio/abc/_eventloop.py
+++ b/src/anyio/abc/_eventloop.py
@@ -3,7 +3,7 @@ from __future__ import annotations
 import math
 import sys
 from abc import ABCMeta, abstractmethod
-from collections.abc import AsyncIterator, Awaitable, Mapping
+from collections.abc import AsyncIterator, Awaitable
 from os import PathLike
 from signal import Signals
 from socket import AddressFamily, SocketKind, socket
@@ -15,6 +15,7 @@ from typing import (
     ContextManager,
     Sequence,
     TypeVar,
+    Union,
     overload,
 )
 
@@ -23,9 +24,12 @@ if sys.version_info >= (3, 11):
 else:
     from typing_extensions import TypeVarTuple, Unpack
 
-if TYPE_CHECKING:
-    from typing import Literal
+if sys.version_info >= (3, 10):
+    from typing import TypeAlias
+else:
+    from typing_extensions import TypeAlias
 
+if TYPE_CHECKING:
     from .._core._synchronization import CapacityLimiter, Event
     from .._core._tasks import CancelScope
     from .._core._testing import TaskInfo
@@ -46,6 +50,7 @@ if TYPE_CHECKING:
 
 T_Retval = TypeVar("T_Retval")
 PosArgsT = TypeVarTuple("PosArgsT")
+StrOrBytesPath: TypeAlias = Union[str, bytes, "PathLike[str]", "PathLike[bytes]"]
 
 
 class AsyncBackend(metaclass=ABCMeta):
@@ -213,51 +218,16 @@ class AsyncBackend(metaclass=ABCMeta):
     def create_blocking_portal(cls) -> BlockingPortal:
         pass
 
-    @classmethod
-    @overload
-    async def open_process(
-        cls,
-        command: str | bytes,
-        *,
-        shell: Literal[True],
-        stdin: int | IO[Any] | None,
-        stdout: int | IO[Any] | None,
-        stderr: int | IO[Any] | None,
-        cwd: str | bytes | PathLike[str] | None = None,
-        env: Mapping[str, str] | None = None,
-        start_new_session: bool = False,
-    ) -> Process:
-        pass
-
-    @classmethod
-    @overload
-    async def open_process(
-        cls,
-        command: Sequence[str | bytes],
-        *,
-        shell: Literal[False],
-        stdin: int | IO[Any] | None,
-        stdout: int | IO[Any] | None,
-        stderr: int | IO[Any] | None,
-        cwd: str | bytes | PathLike[str] | None = None,
-        env: Mapping[str, str] | None = None,
-        start_new_session: bool = False,
-    ) -> Process:
-        pass
-
     @classmethod
     @abstractmethod
     async def open_process(
         cls,
-        command: str | bytes | Sequence[str | bytes],
+        command: StrOrBytesPath | Sequence[StrOrBytesPath],
         *,
-        shell: bool,
         stdin: int | IO[Any] | None,
         stdout: int | IO[Any] | None,
         stderr: int | IO[Any] | None,
-        cwd: str | bytes | PathLike[str] | None = None,
-        env: Mapping[str, str] | None = None,
-        start_new_session: bool = False,
+        **kwargs: Any,
     ) -> Process:
         pass
 


Test patch:
diff --git a/src/anyio/pytest_plugin.py b/src/anyio/pytest_plugin.py
index 1a9f0e3..558c72e 100644
--- a/src/anyio/pytest_plugin.py
+++ b/src/anyio/pytest_plugin.py
@@ -134,6 +134,8 @@ def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
                         if isinstance(exc, (Exit, KeyboardInterrupt, SystemExit)):
                             raise exc from excgrp
 
+                    raise
+
             return True
 
     return None
diff --git a/tests/test_subprocesses.py b/tests/test_subprocesses.py
index 22bf882..84c4b4d 100644
--- a/tests/test_subprocesses.py
+++ b/tests/test_subprocesses.py
@@ -3,12 +3,15 @@ from __future__ import annotations
 import os
 import platform
 import sys
+from collections.abc import Callable
+from contextlib import ExitStack
 from pathlib import Path
 from subprocess import CalledProcessError
 from textwrap import dedent
+from typing import Any
 
 import pytest
-from _pytest.fixtures import FixtureRequest
+from pytest import FixtureRequest
 
 from anyio import CancelScope, ClosedResourceError, open_process, run_process
 from anyio.streams.buffered import BufferedByteReceiveStream
@@ -225,3 +228,64 @@ async def test_process_aexit_cancellation_closes_standard_streams(
 
     with pytest.raises(ClosedResourceError):
         await process.stderr.receive(1)
+
+
+@pytest.mark.parametrize(
+    "argname, argvalue_factory",
+    [
+        pytest.param(
+            "user",
+            lambda: os.getuid(),
+            id="user",
+            marks=[
+                pytest.mark.skipif(
+                    platform.system() == "Windows",
+                    reason="os.getuid() is not available on Windows",
+                )
+            ],
+        ),
+        pytest.param(
+            "group",
+            lambda: os.getgid(),
+            id="user",
+            marks=[
+                pytest.mark.skipif(
+                    platform.system() == "Windows",
+                    reason="os.getgid() is not available on Windows",
+                )
+            ],
+        ),
+        pytest.param("extra_groups", list, id="extra_groups"),
+        pytest.param("umask", lambda: 0, id="umask"),
+    ],
+)
+async def test_py39_arguments(
+    argname: str,
+    argvalue_factory: Callable[[], Any],
+    anyio_backend_name: str,
+    anyio_backend_options: dict[str, Any],
+) -> None:
+    with ExitStack() as stack:
+        if sys.version_info < (3, 9):
+            stack.enter_context(
+                pytest.raises(
+                    TypeError,
+                    match=rf"the {argname!r} argument requires Python 3.9 or later",
+                )
+            )
+
+        try:
+            await run_process(
+                [sys.executable, "-c", "print('hello')"],
+                **{argname: argvalue_factory()},
+            )
+        except TypeError as exc:
+            if (
+                "unexpected keyword argument" in str(exc)
+                and anyio_backend_name == "asyncio"
+                and anyio_backend_options["loop_factory"]
+                and anyio_backend_options["loop_factory"].__module__ == "uvloop"
+            ):
+                pytest.skip(f"the {argname!r} argument is not supported by uvloop yet")
+
+            raise