Add tests for the configuration mutex (47376f4) - derivepassphrase.git

Add tests for the configuration mutex

Marco Ricci commited on 2025-05-18 23:08:44
Zeige 6 geänderte Dateien mit 900 Einfügungen und 35 Löschungen.

Conceptually, we need two types of tests for the configuration mutex:
the first type asserts that the program logic works for *any* kind of
(potentially fake) configuration mutex, and the second type asserts that
the configuration mutex works for *any* set of (potentially fake)
operations.  In both cases, the fake parts usually impose a specific
order of locking or running, controllable from the outside.

We modify the `ConfigurationMutex` class to shift the management of the
lock file descriptor into the lock and unlock functions themselves,
instead of cramming them into `__enter__` and `__exit__`.  (This means
that subclasses need not deal with the lock file, or in fact, with any
of the platform-specific locking machinery.)  A custom subclass in the
`tests` namespace provides a faked mutex that imposes a specific locking
order, controllable from the outside via message passing.  This subclass
is then used with a state machine to test various sets of operations
together with the faked mutex, i.e. to provide tests of the first type.
(The standard tests, executed with the actual mutex implementation,
already provide tests of the second type.)

Regrettably, this set of tests requires concurrency support.
Specifically, because the click testing machinery needs to reset
process-wide properties for each concurrent unit, and because the tests
require this to be done in a time-overlapped manner, the tests
specifically require multiprocessing support.  Care must also be taken
to ensure that coverage management works: that it is enabled in the
subprocesses whenever it is enabled in the parent (which is
a straightforward configuration issue), and that the parent process does
not accidentally remove these coverage measurements as junk files in
case the subprocesses are spawned in specifically prepared and
automatically cleaned up temporary directories (which is a manual labor
issue).

pyproject.toml 7ddc471..ec14389
src/derivepassphrase/_internals/cli_helpers.py 1f3de42..4228a44
src/derivepassphrase/cli.py 51b0224..655c511
tests/__init__.py 653fca6..b578cde
tests/conftest.py 212836f..766fef5
tests/test_derivepassphrase_cli.py d692471..3dd90e3

pyproject.toml

Zeige Datei @ 47376f4

@@ -228,6 +228,8 @@ omit = [
     "__main__.py",
 ]
 dynamic_context = 'test_function'
+concurrency = ['thread', 'multiprocessing']
+sigterm = true
 
 [tool.hatch.build.targets.sdist]
 exclude = [

src/derivepassphrase/_internals/cli_helpers.py

Zeige Datei @ 47376f4

@@ -198,11 +198,13 @@ class ConfigurationMutex:
 
     """
 
-    lock: Callable[[int], None]
-    """A function to lock a given file descriptor exclusively.
+    lock: Callable[[], None]
+    """A function to lock the mutex exclusively.
 
-    On Windows, this uses [`msvcrt.locking`][], on other systems, this
-    uses [`fcntl.flock`][].
+    This implementation uses a file descriptor of a well-known file,
+    which is opened before locking and closed after unlocking (and on
+    error when locking). On Windows, we use [`msvcrt.locking`][], on
+    other systems, we use [`fcntl.flock`][].
 
     Note:
         This is a normal Python function, not a method.
@@ -213,11 +215,14 @@ class ConfigurationMutex:
         [`lock`][] and [`unlock`][] are still compatible.
 
     """
-    unlock: Callable[[int], None]
-    """A function to unlock a given file descriptor.
+    unlock: Callable[[], None]
+    """A function to unlock the mutex.
 
-    On Windows, this uses [`msvcrt.locking`][], on other systems, this
-    uses [`fcntl.flock`][].
+    This implementation uses a file descriptor of a well-known file,
+    which is opened before locking and closed after unlocking (and on
+    error when locking). It will fail if the file descriptor is
+    unavailable. On Windows, we use [`msvcrt.locking`][], on other
+    systems, we use [`fcntl.flock`][].
 
     Note:
         This is a normal Python function, not a method.
@@ -244,11 +249,11 @@ class ConfigurationMutex:
             LK_LOCK = msvcrt.LK_LOCK  # noqa: N806
             LK_UNLCK = msvcrt.LK_UNLCK  # noqa: N806
 
-            def lock_func(fileobj: int) -> None:
-                locking(fileobj, LK_LOCK, LOCK_SIZE)
+            def lock_fd(fd: int, /) -> None:
+                locking(fd, LK_LOCK, LOCK_SIZE)
 
-            def unlock_func(fileobj: int) -> None:
-                locking(fileobj, LK_UNLCK, LOCK_SIZE)
+            def unlock_fd(fd: int, /) -> None:
+                locking(fd, LK_UNLCK, LOCK_SIZE)
 
         else:
             import fcntl  # noqa: PLC0415
@@ -257,11 +262,31 @@ class ConfigurationMutex:
             LOCK_EX = fcntl.LOCK_EX  # noqa: N806
             LOCK_UN = fcntl.LOCK_UN  # noqa: N806
 
-            def lock_func(fileobj: int) -> None:
-                flock(fileobj, LOCK_EX)
+            def lock_fd(fd: int, /) -> None:
+                flock(fd, LOCK_EX)
 
-            def unlock_func(fileobj: int) -> None:
-                flock(fileobj, LOCK_UN)
+            def unlock_fd(fd: int, /) -> None:
+                flock(fd, LOCK_UN)
+
+        def lock_func() -> None:
+            with self.write_lock_condition:
+                self.write_lock_condition.wait_for(
+                    lambda: self.write_lock_fileobj is None
+                )
+                self.write_lock_condition.notify()
+                self.write_lock_file.touch()
+                self.write_lock_fileobj = self.write_lock_file.open('wb')
+                lock_fd(self.write_lock_fileobj.fileno())
+
+        def unlock_func() -> None:
+            with self.write_lock_condition:
+                assert self.write_lock_fileobj is not None, (
+                    'We lost track of the configuration write lock '
+                    'file object, so we cannot unlock it anymore!'
+                )
+                unlock_fd(self.write_lock_fileobj.fileno())
+                self.write_lock_fileobj.close()
+                self.write_lock_fileobj = None
 
         self.lock = lock_func
         self.unlock = unlock_func
@@ -271,13 +296,7 @@ class ConfigurationMutex:
 
     def __enter__(self) -> Self:
         """Enter the context, locking the configuration file."""  # noqa: DOC201
-        with self.write_lock_condition:
-            self.write_lock_condition.wait_for(
-                lambda: self.write_lock_fileobj is None
-            )
-            self.write_lock_file.touch()
-            self.write_lock_fileobj = self.write_lock_file.open('wb')
-            self.lock(self.write_lock_fileobj.fileno())
+        self.lock()
         return self
 
     def __exit__(
@@ -288,14 +307,7 @@ class ConfigurationMutex:
         /,
     ) -> Literal[False]:
         """Exit the context, releasing the lock on the configuration file."""  # noqa: DOC201
-        with self.write_lock_condition:
-            assert self.write_lock_fileobj is not None, (
-                'We lost track of the configuration write lock file object, '
-                'so we cannot unlock it anymore!'
-            )
-            self.unlock(self.write_lock_fileobj.fileno())
-            self.write_lock_fileobj.close()
-            self.write_lock_fileobj = None
+        self.unlock()
         return False
 
 

src/derivepassphrase/cli.py

Zeige Datei @ 47376f4

@@ -36,6 +36,7 @@ from derivepassphrase._internals import cli_messages as _msg
 if TYPE_CHECKING:
     from collections.abc import Sequence
     from collections.abc import Set as AbstractSet
+    from contextlib import AbstractContextManager
 
 __all__ = ('derivepassphrase',)
 
@@ -737,7 +738,10 @@ class _VaultContext:  # noqa: PLR0904
                 )
                 raise click.UsageError(str(err_msg))
 
-    def get_mutex(self, op: str) -> contextlib.AbstractContextManager[None]:
+    def get_mutex(
+        self,
+        op: str,
+    ) -> AbstractContextManager[AbstractContextManager | None]:
         """Return a mutex for accessing the configuration on disk.
 
         The mutex is a context manager, and will lock out other threads

tests/__init__.py

Zeige Datei @ 47376f4

...	...	@@ -16,6 +16,7 @@ import pathlib
16	16	import re
17	17	import shlex
18	18	import stat
	19	+import sys
19	20	import tempfile
20	21	import types
21	22	import zipfile
...	...	@@ -1603,6 +1604,65 @@ available. Usually this means that the test targets the
1603	1604	`derivepassphrase export vault` subcommand, whose functionality depends
1604	1605	on cryptography support being available.
1605	1606	"""
	1607	+skip_if_no_multiprocessing_support = pytest.mark.skipif(
	1608	+ importlib.util.find_spec('multiprocessing') is None,
	1609	+ reason='no "multiprocessing" support',
	1610	+)
	1611	+"""
	1612	+A cached pytest mark to skip this test if multiprocessing support is not
	1613	+available. Usually this means that the test targets the concurrency
	1614	+features of `derivepassphrase`, which is generally only possible to test
	1615	+in separate processes because the testing machinery operates on
	1616	+process-global state.
	1617	+"""
	1618	+
	1619	+MIN_CONCURRENCY = 4
	1620	+"""
	1621	+The minimum amount of concurrent threads used for testing.
	1622	+"""
	1623	+
	1624	+
	1625	+def get_concurrency_limit() -> int:
	1626	+ """Return the imposed limit on the number of concurrent threads.
	1627	+
	1628	+ We use [`os.process_cpu_count`][] as the limit on Python 3.13 and
	1629	+ higher, and [`os.cpu_count`][] on Python 3.12 and below. On
	1630	+ Python 3.12 and below, we explicitly support the `PYTHON_CPU_COUNT`
	1631	+ environment variable. We guarantee at least [`MIN_CONCURRENCY`][]
	1632	+ many threads in any case.
	1633	+
	1634	+ """ # noqa: RUF002
	1635	+ result: int \| None = None
	1636	+ if sys.version_info >= (3, 13):
	1637	+ result = os.process_cpu_count()
	1638	+ else:
	1639	+ try:
	1640	+ cpus = os.sched_getaffinity(os.getpid())
	1641	+ except AttributeError:
	1642	+ pass
	1643	+ else:
	1644	+ result = len(cpus)
	1645	+ return max(result if result is not None else 0, MIN_CONCURRENCY)
	1646	+
	1647	+
	1648	+def get_concurrency_step_count(
	1649	+ settings: hypothesis.settings \| None = None,
	1650	+) -> int:
	1651	+ """Return the desired step count for concurrency-related tests.
	1652	+
	1653	+ This is the smaller of the [general concurrency
	1654	+ limit][tests.get_concurrency_limit] and the step count from the
	1655	+ current hypothesis settings.
	1656	+
	1657	+ Args:
	1658	+ settings:
	1659	+ The hypothesis settings for a specific tests. If not given,
	1660	+ then the current profile will be queried directly.
	1661	+
	1662	+ """
	1663	+ if settings is None: # pragma: no cover
	1664	+ settings = hypothesis.settings()
	1665	+ return min(get_concurrency_limit(), settings.stateful_step_count)
1606	1666
1607	1667
1608	1668	def list_keys(self: Any = None) -> list[_types.SSHKeyCommentPair]:

tests/conftest.py

Zeige Datei @ 47376f4

@@ -111,7 +111,7 @@ _hypothesis_settings_setup()
 
 # https://docs.pytest.org/en/stable/explanation/fixtures.html#a-note-about-fixture-cleanup
 # https://github.com/pytest-dev/pytest/issues/5243#issuecomment-491522595
-@pytest.fixture(scope='session', autouse=True)
+@pytest.fixture(scope='session', autouse=False)
 def term_handler() -> Iterator[None]:  # pragma: no cover
     try:
         import signal  # noqa: PLC0415

tests/test_derivepassphrase_cli.py

Zeige Datei @ 47376f4

@@ -14,6 +14,7 @@ import json
 import logging
 import os
 import pathlib
+import queue
 import re
 import shlex
 import shutil
@@ -22,13 +23,13 @@ import tempfile
 import textwrap
 import types
 import warnings
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 import click.testing
 import hypothesis
 import pytest
 from hypothesis import stateful, strategies
-from typing_extensions import Any, NamedTuple
+from typing_extensions import Any, NamedTuple, TypeAlias
 
 import tests
 from derivepassphrase import _types, cli, ssh_agent, vault
@@ -39,6 +40,7 @@ from derivepassphrase._internals import (
 )
 
 if TYPE_CHECKING:
+    import multiprocessing
     from collections.abc import Callable, Iterable, Iterator, Sequence
     from collections.abc import Set as AbstractSet
     from typing import NoReturn
@@ -6002,6 +6004,791 @@ TestConfigManagement = ConfigManagementStateMachine.TestCase
 """The [`unittest.TestCase`][] class that will actually be run."""
 
 
+class FakeConfigurationMutexAction(NamedTuple):
+    """An action/a step in the [`FakeConfigurationMutexStateMachine`][].
+
+    Attributes:
+        command_line:
+            The command-line for `derivepassphrase vault` to execute.
+        input:
+            The input to this command.
+
+    """
+
+    command_line: list[str]
+    """"""
+    input: str | bytes | None = None
+    """"""
+
+
+def run_actions_handler(
+    id_num: int,
+    action: FakeConfigurationMutexAction,
+    *,
+    input_queue: queue.Queue,
+    output_queue: queue.Queue,
+    timeout: int,
+) -> None:
+    """Prepare the faked mutex, then run `action`.
+
+    This is a top-level handler function -- to be used in a new
+    [`multiprocessing.Process`][] -- to run a single action from the
+    [`FakeConfigurationMutexStateMachine`][].  Output from this function
+    must be sent down the output queue instead of relying on the call
+    stack.  Additionally, because this runs in a separate process, we
+    need to restart coverage tracking if it is currently running.
+
+    Args:
+        id_num:
+            The internal ID of this subprocess.
+        action:
+            The action to execute.
+        input_queue:
+            The queue for data passed from the manager/parent process to
+            this subprocess.
+        output_queue:
+            The queue for data passed from this subprocess to the
+            manager/parent process.
+        timeout:
+            The maximum amount of time to wait for a data transfer along
+            the input or the output queue.  If exceeded, we exit
+            immediately.
+
+    """
+    with pytest.MonkeyPatch.context() as monkeypatch:
+        monkeypatch.setattr(
+            cli_helpers,
+            'configuration_mutex',
+            lambda: FakeConfigurationMutexStateMachine.ConfigurationMutexStub(
+                my_id=id_num,
+                input_queue=input_queue,
+                output_queue=output_queue,
+                timeout=timeout,
+            ),
+        )
+        runner = tests.CliRunner(mix_stderr=False)
+        try:
+            result = runner.invoke(
+                cli.derivepassphrase_vault,
+                args=action.command_line,
+                input=action.input,
+                catch_exceptions=True,
+            )
+            output_queue.put(
+                FakeConfigurationMutexStateMachine.IPCMessage(
+                    id_num,
+                    'result',
+                    (
+                        result.clean_exit(empty_stderr=False),
+                        copy.copy(result.stdout),
+                        copy.copy(result.stderr),
+                    ),
+                ),
+                block=True,
+                timeout=timeout,
+            )
+        except Exception as exc:  # pragma: no cover  # noqa: BLE001
+            output_queue.put(
+                FakeConfigurationMutexStateMachine.IPCMessage(
+                    id_num, 'exception', exc
+                ),
+                block=False,
+            )
+
+
+@hypothesis.settings(
+    stateful_step_count=tests.get_concurrency_step_count(),
+    deadline=None,
+)
+class FakeConfigurationMutexStateMachine(stateful.RuleBasedStateMachine):
+    """A state machine simulating the (faked) configuration mutex.
+
+    Generate an ordered set of concurrent writers to the
+    derivepassphrase configuration, then test that the writers' accesses
+    are serialized correctly, i.e., test that the writers correctly use
+    the mutex to avoid concurrent accesses, under the assumption that
+    the mutex itself is correctly implemented.
+
+    We use a custom mutex implementation to both ensure that all writers
+    attempt to lock the configuration at the same time and that the lock
+    is granted in our desired order.  This test is therefore independent
+    of the actual (operating system-specific) mutex implementation in
+    `derivepassphrase`.
+
+    Attributes:
+        setting:
+            A bundle for single-service settings.
+        configuration:
+            A bundle for full vault configurations.
+
+    """
+
+    class IPCMessage(NamedTuple):
+        """A message for inter-process communication.
+
+        Used by the configuration mutex stub class to affect/signal the
+        control flow amongst the linked mutex clients.
+
+        Attributes:
+            child_id:
+                The ID of the sending or receiving child process.
+            message:
+                One of "ready", "go", "config", "result" or "exception".
+            payload:
+                The (optional) message payload.
+
+        """
+
+        child_id: int
+        """"""
+        message: Literal['ready', 'go', 'config', 'result', 'exception']
+        """"""
+        payload: object | None
+        """"""
+
+    class ConfigurationMutexStub(cli_helpers.ConfigurationMutex):
+        """Configuration mutex subclass that enforces a locking order.
+
+        Each configuration mutex stub object ("mutex client") has an
+        associated ID, and one read-only and one write-only pipe
+        (actually: [`multiprocessing.Queue`][] objects) to the "manager"
+        instance coordinating these stub objects.  First, the mutex
+        client signals readiness, then the manager signals when the
+        mutex shall be considered "acquired", then finally the mutex
+        client sends the result back (simultaneously releasing the mutex
+        again).  The manager may optionally send an abort signal if the
+        operations take too long.
+
+        This subclass also copies the effective vault configuration
+        to `intermediate_configs` upon releasing the lock.
+
+        """
+
+        def __init__(
+            self,
+            *,
+            my_id: int,
+            timeout: int,
+            input_queue: queue.Queue[
+                FakeConfigurationMutexStateMachine.IPCMessage
+            ],
+            output_queue: queue.Queue[
+                FakeConfigurationMutexStateMachine.IPCMessage
+            ],
+        ) -> None:
+            """Initialize this mutex client.
+
+            Args:
+                my_id:
+                    The ID of this client.
+                timeout:
+                    The timeout for each get and put operation on the
+                    queues.
+                input_queue:
+                    The message queue for IPC messages from the manager
+                    instance to this mutex client.
+                output_queue:
+                    The message queue for IPC messages from this mutex
+                    client to the manager instance.
+
+            """
+            super().__init__()
+
+            def lock() -> None:
+                """Simulate locking of the mutex.
+
+                Issue a "ready" message, wait for a "go", then return.
+                If an exception occurs, issue an "exception" message,
+                then raise the exception.
+
+                """
+                IPCMessage: TypeAlias = (
+                    FakeConfigurationMutexStateMachine.IPCMessage
+                )
+                try:
+                    output_queue.put(
+                        IPCMessage(my_id, 'ready', None),
+                        block=True,
+                        timeout=timeout,
+                    )
+                    ok = input_queue.get(block=True, timeout=timeout)
+                    if ok != IPCMessage(my_id, 'go', None):  # pragma: no cover
+                        output_queue.put(
+                            IPCMessage(my_id, 'exception', ok), block=False
+                        )
+                        raise (
+                            ok[2]
+                            if isinstance(ok[2], BaseException)
+                            else RuntimeError(ok[2])
+                        )
+                except (queue.Empty, queue.Full) as exc:  # pragma: no cover
+                    output_queue.put(
+                        IPCMessage(my_id, 'exception', exc), block=False
+                    )
+                    return
+
+            def unlock() -> None:
+                """Simulate unlocking of the mutex.
+
+                Issue a "config" message, then return.  If an exception
+                occurs, issue an "exception" message, then raise the
+                exception.
+
+                """
+                IPCMessage: TypeAlias = (
+                    FakeConfigurationMutexStateMachine.IPCMessage
+                )
+                try:
+                    output_queue.put(
+                        IPCMessage(
+                            my_id,
+                            'config',
+                            copy.copy(cli_helpers.load_config()),
+                        ),
+                        block=True,
+                        timeout=timeout,
+                    )
+                except (queue.Empty, queue.Full) as exc:  # pragma: no cover
+                    output_queue.put(
+                        IPCMessage(my_id, 'exception', exc), block=False
+                    )
+                    raise
+
+            self.lock = lock
+            self.unlock = unlock
+
+    setting: stateful.Bundle[_types.VaultConfigServicesSettings] = (
+        stateful.Bundle('setting')
+    )
+    """"""
+    configuration: stateful.Bundle[_types.VaultConfig] = stateful.Bundle(
+        'configuration'
+    )
+    """"""
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        """Initialize the state machine."""
+        super().__init__(*args, **kwargs)
+        self.actions: list[FakeConfigurationMutexAction] = []
+        # Determine the step count by poking around in the hypothesis
+        # internals. As this isn't guaranteed to be stable, turn off
+        # coverage.
+        try:  # pragma: no cover
+            settings: hypothesis.settings | None
+            settings = FakeConfigurationMutexStateMachine.TestCase.settings
+        except AttributeError:  # pragma: no cover
+            settings = None
+        self.step_count = tests.get_concurrency_step_count(settings)
+
+    @stateful.initialize(
+        target=configuration,
+        configs=strategies.lists(
+            vault_full_config(),
+            min_size=8,
+            max_size=8,
+        ),
+    )
+    def declare_initial_configs(
+        self,
+        configs: list[_types.VaultConfig],
+    ) -> stateful.MultipleResults[_types.VaultConfig]:
+        """Initialize the configuration bundle with eight configurations."""
+        return stateful.multiple(*configs)
+
+    @stateful.initialize(
+        target=setting,
+        configs=strategies.lists(
+            vault_full_config(),
+            min_size=4,
+            max_size=4,
+        ),
+    )
+    def extract_initial_settings(
+        self,
+        configs: list[_types.VaultConfig],
+    ) -> stateful.MultipleResults[_types.VaultConfigServicesSettings]:
+        """Initialize the settings bundle with four service settings."""
+        settings: list[_types.VaultConfigServicesSettings] = []
+        for c in configs:
+            settings.extend(c['services'].values())
+        return stateful.multiple(*map(copy.deepcopy, settings))
+
+    @stateful.initialize(
+        config=vault_full_config(),
+    )
+    def declare_initial_action(
+        self,
+        config: _types.VaultConfig,
+    ) -> None:
+        """Initialize the actions bundle from the configuration bundle.
+
+        This is roughly comparable to the
+        [`add_import_configuration_action`][] general rule, but adding
+        it as a separate initialize rule avoids having to guard every
+        other action-amending rule against empty action sequences, which
+        would discard huge portions of the rule selection search space
+        and thus trigger loads of hypothesis health check warnings.
+
+        """
+        command_line = ['--import', '-', '--overwrite-existing']
+        input = json.dumps(config)  # noqa: A001
+        hypothesis.note(f'# {command_line = }, {input = }')
+        action = FakeConfigurationMutexAction(
+            command_line=command_line, input=input
+        )
+        self.actions.append(action)
+
+    @stateful.rule(
+        setting=setting.filter(bool),
+        maybe_unset=strategies.sets(
+            strategies.sampled_from(VALID_PROPERTIES),
+            max_size=3,
+        ),
+        overwrite=strategies.booleans(),
+    )
+    def add_set_globals_action(
+        self,
+        setting: _types.VaultConfigGlobalSettings,
+        maybe_unset: set[str],
+        overwrite: bool,
+    ) -> None:
+        """Set the global settings of a configuration.
+
+        Args:
+            setting:
+                The new global settings.
+            maybe_unset:
+                Settings keys to additionally unset, if not already
+                present in the new settings.  Corresponds to the
+                `--unset` command-line argument.
+            overwrite:
+                Overwrite the settings object if true, or merge if
+                false.  Corresponds to the `--overwrite-existing` and
+                `--merge-existing` command-line arguments.
+
+        """
+        maybe_unset = set(maybe_unset) - setting.keys()
+        command_line = (
+            [
+                '--config',
+                '--overwrite-existing' if overwrite else '--merge-existing',
+            ]
+            + [f'--unset={key}' for key in maybe_unset]
+            + [
+                f'--{key}={value}'
+                for key, value in setting.items()
+                if key in VALID_PROPERTIES
+            ]
+        )
+        input = None  # noqa: A001
+        hypothesis.note(f'# {command_line = }, {input = }')
+        action = FakeConfigurationMutexAction(
+            command_line=command_line, input=input
+        )
+        self.actions.append(action)
+
+    @stateful.rule(
+        service=strategies.sampled_from(KNOWN_SERVICES),
+        setting=setting.filter(bool),
+        maybe_unset=strategies.sets(
+            strategies.sampled_from(VALID_PROPERTIES),
+            max_size=3,
+        ),
+        overwrite=strategies.booleans(),
+    )
+    def add_set_service_action(
+        self,
+        service: str,
+        setting: _types.VaultConfigServicesSettings,
+        maybe_unset: set[str],
+        overwrite: bool,
+    ) -> None:
+        """Set the named service settings for a configuration.
+
+        Args:
+            service:
+                The name of the service to set.
+            setting:
+                The new service settings.
+            maybe_unset:
+                Settings keys to additionally unset, if not already
+                present in the new settings.  Corresponds to the
+                `--unset` command-line argument.
+            overwrite:
+                Overwrite the settings object if true, or merge if
+                false.  Corresponds to the `--overwrite-existing` and
+                `--merge-existing` command-line arguments.
+
+        """
+        maybe_unset = set(maybe_unset) - setting.keys()
+        command_line = (
+            [
+                '--config',
+                '--overwrite-existing' if overwrite else '--merge-existing',
+            ]
+            + [f'--unset={key}' for key in maybe_unset]
+            + [
+                f'--{key}={value}'
+                for key, value in setting.items()
+                if key in VALID_PROPERTIES
+            ]
+            + ['--', service]
+        )
+        input = None  # noqa: A001
+        hypothesis.note(f'# {command_line = }, {input = }')
+        action = FakeConfigurationMutexAction(
+            command_line=command_line, input=input
+        )
+        self.actions.append(action)
+
+    @stateful.rule()
+    def add_purge_global_action(
+        self,
+    ) -> None:
+        """Purge the globals of a configuration."""
+        command_line = ['--delete-globals']
+        input = None  # 'y'  # noqa: A001
+        hypothesis.note(f'# {command_line = }, {input = }')
+        action = FakeConfigurationMutexAction(
+            command_line=command_line, input=input
+        )
+        self.actions.append(action)
+
+    @stateful.rule(
+        service=strategies.sampled_from(KNOWN_SERVICES),
+    )
+    def add_purge_service_action(
+        self,
+        service: str,
+    ) -> None:
+        """Purge the settings of a named service in a configuration.
+
+        Args:
+            service:
+                The service name to purge.
+
+        """
+        command_line = ['--delete', '--', service]
+        input = None  # 'y'  # noqa: A001
+        hypothesis.note(f'# {command_line = }, {input = }')
+        action = FakeConfigurationMutexAction(
+            command_line=command_line, input=input
+        )
+        self.actions.append(action)
+
+    @stateful.rule()
+    def add_purge_all_action(
+        self,
+    ) -> None:
+        """Purge the entire configuration."""
+        command_line = ['--clear']
+        input = None  # 'y'  # noqa: A001
+        hypothesis.note(f'# {command_line = }, {input = }')
+        action = FakeConfigurationMutexAction(
+            command_line=command_line, input=input
+        )
+        self.actions.append(action)
+
+    @stateful.rule(
+        config_to_import=configuration,
+        overwrite=strategies.booleans(),
+    )
+    def add_import_configuration_action(
+        self,
+        config_to_import: _types.VaultConfig,
+        overwrite: bool,
+    ) -> None:
+        """Import the given configuration.
+
+        Args:
+            config_to_import:
+                The configuration to import.
+            overwrite:
+                Overwrite the base configuration if true, or merge if
+                false.  Corresponds to the `--overwrite-existing` and
+                `--merge-existing` command-line arguments.
+
+        """
+        command_line = ['--import', '-'] + (
+            ['--overwrite-existing'] if overwrite else []
+        )
+        input = json.dumps(config_to_import)  # noqa: A001
+        hypothesis.note(f'# {command_line = }, {input = }')
+        action = FakeConfigurationMutexAction(
+            command_line=command_line, input=input
+        )
+        self.actions.append(action)
+
+    @stateful.precondition(lambda self: len(self.actions) > 0)
+    @stateful.invariant()
+    def run_actions(  # noqa: C901
+        self,
+    ) -> None:
+        """Run the actions, serially and concurrently.
+
+        Run the actions once serially, then once more concurrently with
+        the faked configuration mutex, and assert that both runs yield
+        identical intermediate and final results.
+
+        We must run the concurrent version in processes, not threads or
+        Python async functions, because the `click` testing machinery
+        manipulates global properties (e.g. the standard I/O streams,
+        the current directory, and the environment), and we require this
+        manipulation to happen in a time-overlapped manner.
+
+        However, running multiple processes increases the risk of the
+        operating system imposing process count or memory limits on us.
+        We therefore skip the test as a whole if we fail to start a new
+        process due to lack of necessary resources (memory, processes,
+        or open file descriptors).
+
+        """
+        if not TYPE_CHECKING:  # pragma: no branch
+            multiprocessing = pytest.importorskip('multiprocessing')
+        IPCMessage: TypeAlias = FakeConfigurationMutexStateMachine.IPCMessage
+        intermediate_configs: dict[int, _types.VaultConfig] = {}
+        intermediate_results: dict[
+            int, tuple[bool, str | None, str | None]
+        ] = {}
+        true_configs: dict[int, _types.VaultConfig] = {}
+        true_results: dict[int, tuple[bool, str | None, str | None]] = {}
+        timeout = 5
+        actions = self.actions
+        mp = multiprocessing.get_context()
+        # Coverage tracking writes coverage data to the current working
+        # directory, but because the subprocesses are spawned within the
+        # `tests.isolated_vault_config` context manager, their starting
+        # working directory is the isolated one, not the original one.
+        orig_cwd = pathlib.Path.cwd()
+
+        fatal_process_creation_errnos = {
+            # Specified by POSIX for fork(3).
+            errno.ENOMEM,
+            # Specified by POSIX for fork(3).
+            errno.EAGAIN,
+            # Specified by Linux/glibc for fork(3)
+            getattr(errno, 'ENOSYS', errno.ENOMEM),
+            # Specified by POSIX for posix_spawn(3).
+            errno.EINVAL,
+        }
+
+        hypothesis.note(f'# {actions = }')
+
+        stack = contextlib.ExitStack()
+        with stack:
+            runner = tests.CliRunner(mix_stderr=False)
+            monkeypatch = stack.enter_context(pytest.MonkeyPatch.context())
+            stack.enter_context(
+                tests.isolated_vault_config(
+                    monkeypatch=monkeypatch,
+                    runner=runner,
+                    vault_config={'services': {}},
+                )
+            )
+            for i, action in enumerate(actions):
+                result = runner.invoke(
+                    cli.derivepassphrase_vault,
+                    args=action.command_line,
+                    input=action.input,
+                    catch_exceptions=True,
+                )
+                true_configs[i] = copy.copy(cli_helpers.load_config())
+                true_results[i] = (
+                    result.clean_exit(empty_stderr=False),
+                    result.stdout,
+                    result.stderr,
+                )
+
+        with stack:  # noqa: PLR1702
+            runner = tests.CliRunner(mix_stderr=False)
+            monkeypatch = stack.enter_context(pytest.MonkeyPatch.context())
+            stack.enter_context(
+                tests.isolated_vault_config(
+                    monkeypatch=monkeypatch,
+                    runner=runner,
+                    vault_config={'services': {}},
+                )
+            )
+
+            child_output_queue: multiprocessing.Queue[IPCMessage] = mp.Queue()
+            child_input_queues: list[
+                multiprocessing.Queue[IPCMessage] | None
+            ] = []
+            processes: list[multiprocessing.process.BaseProcess] = []
+            processes_pending: set[multiprocessing.process.BaseProcess] = set()
+            ready_wait: set[int] = set()
+
+            try:
+                for i, action in enumerate(actions):
+                    q: multiprocessing.Queue[IPCMessage] | None = mp.Queue()
+                    try:
+                        p: multiprocessing.process.BaseProcess = mp.Process(
+                            name=f'fake-mutex-action-{i:02d}',
+                            target=run_actions_handler,
+                            kwargs={
+                                'id_num': i,
+                                'timeout': timeout,
+                                'action': action,
+                                'input_queue': q,
+                                'output_queue': child_output_queue,
+                            },
+                            daemon=False,
+                        )
+                        p.start()
+                    except OSError as exc:  # pragma: no cover
+                        if exc.errno in fatal_process_creation_errnos:
+                            pytest.skip(
+                                'cannot test mutex functionality due to '
+                                'lack of system resources for '
+                                'creating enough subprocesses'
+                            )
+                        raise
+                    else:
+                        processes.append(p)
+                        processes_pending.add(p)
+                        child_input_queues.append(q)
+                        ready_wait.add(i)
+
+                while processes_pending:
+                    try:
+                        self.mainloop(
+                            timeout=timeout,
+                            child_output_queue=child_output_queue,
+                            child_input_queues=child_input_queues,
+                            ready_wait=ready_wait,
+                            intermediate_configs=intermediate_configs,
+                            intermediate_results=intermediate_results,
+                            processes=processes,
+                            processes_pending=processes_pending,
+                            block=True,
+                        )
+                    except Exception as exc:  # pragma: no cover
+                        for i, q in enumerate(child_input_queues):
+                            if q:
+                                q.put(IPCMessage(i, 'exception', exc))
+                        for p in processes_pending:
+                            p.join(timeout=timeout)
+                        raise
+            finally:
+                try:
+                    while True:
+                        try:
+                            self.mainloop(
+                                timeout=timeout,
+                                child_output_queue=child_output_queue,
+                                child_input_queues=child_input_queues,
+                                ready_wait=ready_wait,
+                                intermediate_configs=intermediate_configs,
+                                intermediate_results=intermediate_results,
+                                processes=processes,
+                                processes_pending=processes_pending,
+                                block=False,
+                            )
+                        except queue.Empty:
+                            break
+                finally:
+                    # The subprocesses have this
+                    # `tests.isolated_vault_config` directory as their
+                    # startup and working directory, so systems like
+                    # coverage tracking write their data files to this
+                    # directory.  We need to manually move them back to
+                    # the starting working directory if they are to
+                    # survive this test.
+                    for coverage_file in pathlib.Path.cwd().glob(
+                        '.coverage.*'
+                    ):
+                        shutil.move(coverage_file, orig_cwd)
+        hypothesis.note(
+            f'# {true_results = }, {intermediate_results = }, '
+            f'identical = {true_results == intermediate_results}'
+        )
+        hypothesis.note(
+            f'# {true_configs = }, {intermediate_configs = }, '
+            f'identical = {true_configs == intermediate_configs}'
+        )
+        assert intermediate_results == true_results
+        assert intermediate_configs == true_configs
+
+    @staticmethod
+    def mainloop(
+        *,
+        timeout: int,
+        child_output_queue: multiprocessing.Queue[
+            FakeConfigurationMutexStateMachine.IPCMessage
+        ],
+        child_input_queues: list[
+            multiprocessing.Queue[
+                FakeConfigurationMutexStateMachine.IPCMessage
+            ]
+            | None
+        ],
+        ready_wait: set[int],
+        intermediate_configs: dict[int, _types.VaultConfig],
+        intermediate_results: dict[int, tuple[bool, str | None, str | None]],
+        processes: list[multiprocessing.process.BaseProcess],
+        processes_pending: set[multiprocessing.process.BaseProcess],
+        block: bool = True,
+    ) -> None:
+        IPCMessage: TypeAlias = FakeConfigurationMutexStateMachine.IPCMessage
+        msg = child_output_queue.get(block=block, timeout=timeout)
+        # TODO(the-13th-letter): Rewrite using structural pattern
+        # matching.
+        # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
+        if (  # pragma: no cover
+            isinstance(msg, IPCMessage)
+            and msg[1] == 'exception'
+            and isinstance(msg[2], Exception)
+        ):
+            e = msg[2]
+            raise e
+        if isinstance(msg, IPCMessage) and msg[1] == 'ready':
+            n = msg[0]
+            ready_wait.remove(n)
+            if not ready_wait:
+                assert child_input_queues
+                assert child_input_queues[0]
+                child_input_queues[0].put(
+                    IPCMessage(0, 'go', None),
+                    block=True,
+                    timeout=timeout,
+                )
+        elif isinstance(msg, IPCMessage) and msg[1] == 'config':
+            n = msg[0]
+            config = msg[2]
+            intermediate_configs[n] = cast('_types.VaultConfig', config)
+        elif isinstance(msg, IPCMessage) and msg[1] == 'result':
+            n = msg[0]
+            result_ = msg[2]
+            result_tuple: tuple[bool, str | None, str | None] = cast(
+                'tuple[bool, str | None, str | None]', result_
+            )
+            intermediate_results[n] = result_tuple
+            child_input_queues[n] = None
+            p = processes[n]
+            p.join(timeout=timeout)
+            assert not p.is_alive()
+            processes_pending.remove(p)
+            assert result_tuple[0], (
+                f'action #{n} exited with an error: {result_tuple!r}'
+            )
+            if n + 1 < len(processes):
+                next_child_input_queue = child_input_queues[n + 1]
+                assert next_child_input_queue
+                next_child_input_queue.put(
+                    IPCMessage(n + 1, 'go', None),
+                    block=True,
+                    timeout=timeout,
+                )
+        else:
+            raise AssertionError()
+
+
+TestFakedConfigurationMutex = tests.skip_if_no_multiprocessing_support(
+    FakeConfigurationMutexStateMachine.TestCase
+)
+"""The [`unittest.TestCase`][] class that will actually be run."""
+
+
 def completion_item(
     item: str | click.shell_completion.CompletionItem,
 ) -> click.shell_completion.CompletionItem:


...	...	@@ -228,6 +228,8 @@ omit = [
228	228	"__main__.py",
229	229	]
230	230	dynamic_context = 'test_function'
	231	+concurrency = ['thread', 'multiprocessing']
	232	+sigterm = true
231	233
232	234	[tool.hatch.build.targets.sdist]
233	235	exclude = [

...	...	@@ -198,11 +198,13 @@ class ConfigurationMutex:
198	198
199	199	"""
200	200
201		- lock: Callable[[int], None]
202		- """A function to lock a given file descriptor exclusively.
	201	+ lock: Callable[[], None]
	202	+ """A function to lock the mutex exclusively.
203	203
204		- On Windows, this uses [`msvcrt.locking`][], on other systems, this
205		- uses [`fcntl.flock`][].
	204	+ This implementation uses a file descriptor of a well-known file,
	205	+ which is opened before locking and closed after unlocking (and on
	206	+ error when locking). On Windows, we use [`msvcrt.locking`][], on
	207	+ other systems, we use [`fcntl.flock`][].
206	208
207	209	Note:
208	210	This is a normal Python function, not a method.
...	...	@@ -213,11 +215,14 @@ class ConfigurationMutex:
213	215	[`lock`][] and [`unlock`][] are still compatible.
214	216
215	217	"""
216		- unlock: Callable[[int], None]
217		- """A function to unlock a given file descriptor.
	218	+ unlock: Callable[[], None]
	219	+ """A function to unlock the mutex.
218	220
219		- On Windows, this uses [`msvcrt.locking`][], on other systems, this
220		- uses [`fcntl.flock`][].
	221	+ This implementation uses a file descriptor of a well-known file,
	222	+ which is opened before locking and closed after unlocking (and on
	223	+ error when locking). It will fail if the file descriptor is
	224	+ unavailable. On Windows, we use [`msvcrt.locking`][], on other
	225	+ systems, we use [`fcntl.flock`][].
221	226
222	227	Note:
223	228	This is a normal Python function, not a method.
...	...	@@ -244,11 +249,11 @@ class ConfigurationMutex:
244	249	LK_LOCK = msvcrt.LK_LOCK # noqa: N806
245	250	LK_UNLCK = msvcrt.LK_UNLCK # noqa: N806
246	251
247		- def lock_func(fileobj: int) -> None:
248		- locking(fileobj, LK_LOCK, LOCK_SIZE)
	252	+ def lock_fd(fd: int, /) -> None:
	253	+ locking(fd, LK_LOCK, LOCK_SIZE)
249	254
250		- def unlock_func(fileobj: int) -> None:
251		- locking(fileobj, LK_UNLCK, LOCK_SIZE)
	255	+ def unlock_fd(fd: int, /) -> None:
	256	+ locking(fd, LK_UNLCK, LOCK_SIZE)
252	257
253	258	else:
254	259	import fcntl # noqa: PLC0415
...	...	@@ -257,11 +262,31 @@ class ConfigurationMutex:
257	262	LOCK_EX = fcntl.LOCK_EX # noqa: N806
258	263	LOCK_UN = fcntl.LOCK_UN # noqa: N806
259	264
260		- def lock_func(fileobj: int) -> None:
261		- flock(fileobj, LOCK_EX)
	265	+ def lock_fd(fd: int, /) -> None:
	266	+ flock(fd, LOCK_EX)
262	267
263		- def unlock_func(fileobj: int) -> None:
264		- flock(fileobj, LOCK_UN)
	268	+ def unlock_fd(fd: int, /) -> None:
	269	+ flock(fd, LOCK_UN)
	270	+
	271	+ def lock_func() -> None:
	272	+ with self.write_lock_condition:
	273	+ self.write_lock_condition.wait_for(
	274	+ lambda: self.write_lock_fileobj is None
	275	+ )
	276	+ self.write_lock_condition.notify()
	277	+ self.write_lock_file.touch()
	278	+ self.write_lock_fileobj = self.write_lock_file.open('wb')
	279	+ lock_fd(self.write_lock_fileobj.fileno())
	280	+
	281	+ def unlock_func() -> None:
	282	+ with self.write_lock_condition:
	283	+ assert self.write_lock_fileobj is not None, (
	284	+ 'We lost track of the configuration write lock '
	285	+ 'file object, so we cannot unlock it anymore!'
	286	+ )
	287	+ unlock_fd(self.write_lock_fileobj.fileno())
	288	+ self.write_lock_fileobj.close()
	289	+ self.write_lock_fileobj = None
265	290
266	291	self.lock = lock_func
267	292	self.unlock = unlock_func
...	...	@@ -271,13 +296,7 @@ class ConfigurationMutex:
271	296
272	297	def __enter__(self) -> Self:
273	298	"""Enter the context, locking the configuration file.""" # noqa: DOC201
274		- with self.write_lock_condition:
275		- self.write_lock_condition.wait_for(
276		- lambda: self.write_lock_fileobj is None
277		- )
278		- self.write_lock_file.touch()
279		- self.write_lock_fileobj = self.write_lock_file.open('wb')
280		- self.lock(self.write_lock_fileobj.fileno())
	299	+ self.lock()
281	300	return self
282	301
283	302	def __exit__(
...	...	@@ -288,14 +307,7 @@ class ConfigurationMutex:
288	307	/,
289	308	) -> Literal[False]:
290	309	"""Exit the context, releasing the lock on the configuration file.""" # noqa: DOC201
291		- with self.write_lock_condition:
292		- assert self.write_lock_fileobj is not None, (
293		- 'We lost track of the configuration write lock file object, '
294		- 'so we cannot unlock it anymore!'
295		- )
296		- self.unlock(self.write_lock_fileobj.fileno())
297		- self.write_lock_fileobj.close()
298		- self.write_lock_fileobj = None
	310	+ self.unlock()
299	311	return False
300	312
301	313

...	...	@@ -36,6 +36,7 @@ from derivepassphrase._internals import cli_messages as _msg
36	36	if TYPE_CHECKING:
37	37	from collections.abc import Sequence
38	38	from collections.abc import Set as AbstractSet
	39	+ from contextlib import AbstractContextManager
39	40
40	41	__all__ = ('derivepassphrase',)
41	42
...	...	@@ -737,7 +738,10 @@ class _VaultContext: # noqa: PLR0904
737	738	)
738	739	raise click.UsageError(str(err_msg))
739	740
740		- def get_mutex(self, op: str) -> contextlib.AbstractContextManager[None]:
	741	+ def get_mutex(
	742	+ self,
	743	+ op: str,
	744	+ ) -> AbstractContextManager[AbstractContextManager \| None]:
741	745	"""Return a mutex for accessing the configuration on disk.
742	746
743	747	The mutex is a context manager, and will lock out other threads

...	...	@@ -111,7 +111,7 @@ _hypothesis_settings_setup()
111	111
112	112	# https://docs.pytest.org/en/stable/explanation/fixtures.html#a-note-about-fixture-cleanup
113	113	# https://github.com/pytest-dev/pytest/issues/5243#issuecomment-491522595
114		-@pytest.fixture(scope='session', autouse=True)
	114	+@pytest.fixture(scope='session', autouse=False)
115	115	def term_handler() -> Iterator[None]: # pragma: no cover
116	116	try:
117	117	import signal # noqa: PLC0415