Class AuditedLock

tests/test_fork_coordination.py:207–215

    event: Literal["os.fork", "os.forkpty", "_posixsubprocess.fork_exec"],
) -> None:
    class AuditedLock(BaseFileLock):
        def _acquire(self) -> None:
            self._context.lock_file_fd = None
            if event == "_posixsubprocess.fork_exec":
                sys.audit(event, (), (), None)
            else:
                sys.audit(event)

        _release = _acquire

    with pytest.raises(RuntimeError, match="unsafe while filelock is changing descriptor ownership"):
        AuditedLock(str(tmp_path / "audit.lock"), is_singleton=False).acquire(timeout=0)