added multiuser support and associated tests (#192)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

docs/changelog.rst

@@ -1,5 +1,9 @@
 Changelog
 =========
+v3.10.0 (2023-03-15)
+-------------------
+- Add support for explicit file modes for lockfiles :pr:`192 - by :user:`jahrules`.
+
 v3.9.1 (2023-03-14)
 -------------------
 - Use ``time.perf_counter`` instead of ``time.monotonic`` for calculating timeouts.

src/filelock/_api.py

@@ -39,14 +39,20 @@ def __exit__(
 class BaseFileLock(ABC, contextlib.ContextDecorator):
     """Abstract base class for a file lock object."""
 
-    def __init__(self, lock_file: str | os.PathLike[Any], timeout: float = -1) -> None:
+    def __init__(
+        self,
+        lock_file: str | os.PathLike[Any],
+        timeout: float = -1,
+        mode: int = 0o644,
+    ) -> None:
         """
         Create a new lock object.
 
         :param lock_file: path to the file
         :param timeout: default timeout when acquiring the lock, in seconds. It will be used as fallback value in
         the acquire method, if no timeout value (``None``) is given. If you want to disable the timeout, set it
         to a negative value. A timeout of 0 means, that there is exactly one attempt to acquire the file lock.
+        : param mode: file permissions for the lockfile.
         """
         # The path to the lock file.
         self._lock_file: str = os.fspath(lock_file)
@@ -58,6 +64,9 @@ def __init__(self, lock_file: str | os.PathLike[Any], timeout: float = -1) -> No
         # The default timeout value.
         self._timeout: float = timeout
 
+        # The mode for the lock files
+        self._mode: int = mode
+
         # We use this lock primarily for the lock counter.
         self._thread_lock: Lock = Lock()
 
@@ -170,8 +179,11 @@ def acquire(
                 with self._thread_lock:
                     if not self.is_locked:
                         _LOGGER.debug("Attempting to acquire lock %s on %s", lock_id, lock_filename)
-                        self._acquire()
-
+                        previous_umask = os.umask(0)
+                        try:
+                            self._acquire()
+                        finally:
+                            os.umask(previous_umask)  # reset umask to initial value
                 if self.is_locked:
                     _LOGGER.debug("Lock %s acquired on %s", lock_id, lock_filename)
                     break

src/filelock/_soft.py

@@ -14,14 +14,14 @@ class SoftFileLock(BaseFileLock):
     def _acquire(self) -> None:
         raise_on_exist_ro_file(self._lock_file)
         # first check for exists and read-only mode as the open will mask this case as EEXIST
-        mode = (
+        flags = (
             os.O_WRONLY  # open for writing only
             | os.O_CREAT
             | os.O_EXCL  # together with above raise EEXIST if the file specified by filename exists
             | os.O_TRUNC  # truncate the file to zero byte
         )
         try:
-            fd = os.open(self._lock_file, mode)
+            fd = os.open(self._lock_file, flags, self._mode)
         except OSError as exception:
             if exception.errno == EEXIST:  # expected if cannot lock
                 pass

src/filelock/_unix.py

@@ -31,8 +31,8 @@ class UnixFileLock(BaseFileLock):
         """Uses the :func:`fcntl.flock` to hard lock the lock file on unix systems."""
 
         def _acquire(self) -> None:
-            open_mode = os.O_RDWR | os.O_CREAT | os.O_TRUNC
-            fd = os.open(self._lock_file, open_mode)
+            open_flags = os.O_RDWR | os.O_CREAT | os.O_TRUNC
+            fd = os.open(self._lock_file, open_flags, self._mode)
             try:
                 fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
             except OSError:

src/filelock/_windows.py

@@ -16,13 +16,13 @@ class WindowsFileLock(BaseFileLock):
 
         def _acquire(self) -> None:
             raise_on_exist_ro_file(self._lock_file)
-            mode = (
+            flags = (
                 os.O_RDWR  # open for read and write
                 | os.O_CREAT  # create file if not exists
                 | os.O_TRUNC  # truncate file  if not empty
             )
             try:
-                fd = os.open(self._lock_file, mode)
+                fd = os.open(self._lock_file, flags, self._mode)
             except OSError as exception:
                 if exception.errno == ENOENT:  # No such file or directory
                     raise

tests/test_filelock.py

@@ -2,12 +2,13 @@
 
 import inspect
 import logging
+import os
 import sys
 import threading
 from contextlib import contextmanager
 from inspect import getframeinfo, stack
 from pathlib import Path, PurePath
-from stat import S_IWGRP, S_IWOTH, S_IWUSR
+from stat import S_IWGRP, S_IWOTH, S_IWUSR, filemode
 from types import TracebackType
 from typing import Callable, Iterator, Tuple, Type, Union
 
@@ -418,6 +419,66 @@ def decorated_method() -> None:
     assert not lock.is_locked
 
 
+def test_lock_mode(tmp_path: Path) -> None:
+    lock_path = tmp_path / "a.lock"
+    lock = FileLock(str(lock_path), mode=0o666)
+
+    lock.acquire()
+    assert lock.is_locked
+
+    mode = filemode(os.stat(lock_path).st_mode)
+    assert mode == "-rw-rw-rw-"
+
+    lock.release()
+
+
+def test_lock_mode_soft(tmp_path: Path) -> None:
+    lock_path = tmp_path / "a.lock"
+    lock = SoftFileLock(str(lock_path), mode=0o666)
+
+    lock.acquire()
+    assert lock.is_locked
+
+    mode = filemode(os.stat(lock_path).st_mode)
+    assert mode == "-rw-rw-rw-"
+
+    lock.release()
+
+
+def test_umask(tmp_path: Path) -> None:
+    lock_path = tmp_path / "a.lock"
+    lock = FileLock(str(lock_path), mode=0o666)
+
+    initial_umask = os.umask(0)
+    os.umask(initial_umask)
+
+    lock.acquire()
+    assert lock.is_locked
+
+    current_umask = os.umask(0)
+    os.umask(current_umask)
+    assert initial_umask == current_umask
+
+    lock.release()
+
+
+def test_umask_soft(tmp_path: Path) -> None:
+    lock_path = tmp_path / "a.lock"
+    lock = SoftFileLock(str(lock_path), mode=0o666)
+
+    initial_umask = os.umask(0)
+    os.umask(initial_umask)
+
+    lock.acquire()
+    assert lock.is_locked
+
+    current_umask = os.umask(0)
+    os.umask(current_umask)
+    assert initial_umask == current_umask
+
+    lock.release()
+
+
 def test_wrong_platform(tmp_path: Path) -> None:
     assert not inspect.isabstract(UnixFileLock)
     assert not inspect.isabstract(WindowsFileLock)

whitelist.txt

@@ -5,6 +5,7 @@ caplog
 eacces
 extlinks
 filelock
+filemode
 frameinfo
 fspath
 getframeinfo