Fix lock hang on Windows (#231)

Co-authored-by: TheMatt2 <[email protected]>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Bernát Gábor <[email protected]>

Author: 3 people

.pre-commit-config.yaml

@@ -54,7 +54,7 @@ repos:
       - id: flake8
         additional_dependencies:
           - flake8-bugbear==23.3.23
-          - flake8-comprehensions==3.11.1
+          - flake8-comprehensions==3.12
           - flake8-pytest-style==1.7.2
           - flake8-spellcheck==0.28
           - flake8-unused-arguments==0.0.13

pyproject.toml

@@ -37,13 +37,13 @@ dynamic = [
 optional-dependencies.docs = [
   "furo>=2023.3.27",
   "sphinx>=6.1.3",
-  "sphinx-autodoc-typehints!=1.23.4,>=1.22",
+  "sphinx-autodoc-typehints!=1.23.4,>=1.23",
 ]
 optional-dependencies.testing = [
   "covdefaults>=2.3",
   "coverage>=7.2.3",
   "diff-cover>=7.5",
-  "pytest>=7.2.2",
+  "pytest>=7.3.1",
   "pytest-cov>=4",
   "pytest-mock>=3.10",
   "pytest-timeout>=2.1",

src/filelock/_soft.py

@@ -5,14 +5,14 @@
 from errno import EACCES, EEXIST
 
 from ._api import BaseFileLock
-from ._util import raise_on_exist_ro_file
+from ._util import raise_on_not_writable_file
 
 
 class SoftFileLock(BaseFileLock):
     """Simply watches the existence of the lock file."""
 
     def _acquire(self) -> None:
-        raise_on_exist_ro_file(self._lock_file)
+        raise_on_not_writable_file(self._lock_file)
         # first check for exists and read-only mode as the open will mask this case as EEXIST
         flags = (
             os.O_WRONLY  # open for writing only

src/filelock/_util.py

@@ -2,19 +2,36 @@
 
 import os
 import stat
+import sys
+from errno import EACCES, EISDIR
 
 
-def raise_on_exist_ro_file(filename: str) -> None:
+def raise_on_not_writable_file(filename: str) -> None:
+    """
+    Raise an exception if attempting to open the file for writing would fail.
+    This is done so files that will never be writable can be separated from
+    files that are writable but currently locked
+    :param filename: file to check
+    :raises OSError: as if the file was opened for writing
+    """
     try:
         file_stat = os.stat(filename)  # use stat to do exists + can write to check without race condition
     except OSError:
         return None  # swallow does not exist or other errors
 
     if file_stat.st_mtime != 0:  # if os.stat returns but modification is zero that's an invalid os.stat - ignore it
         if not (file_stat.st_mode & stat.S_IWUSR):
-            raise PermissionError(f"Permission denied: {filename!r}")
+            raise PermissionError(EACCES, "Permission denied", filename)
+
+        if stat.S_ISDIR(file_stat.st_mode):
+            if sys.platform == "win32":  # pragma: win32 cover
+                # On Windows, this is PermissionError
+                raise PermissionError(EACCES, "Permission denied", filename)
+            else:  # pragma: win32 no cover
+                # On linux / macOS, this is IsADirectoryError
+                raise IsADirectoryError(EISDIR, "Is a directory", filename)
 
 
 __all__ = [
-    "raise_on_exist_ro_file",
+    "raise_on_not_writable_file",
 ]

src/filelock/_windows.py

@@ -2,11 +2,11 @@
 
 import os
 import sys
-from errno import ENOENT
+from errno import EACCES
 from typing import cast
 
 from ._api import BaseFileLock
-from ._util import raise_on_exist_ro_file
+from ._util import raise_on_not_writable_file
 
 if sys.platform == "win32":  # pragma: win32 cover
     import msvcrt
@@ -15,22 +15,24 @@ class WindowsFileLock(BaseFileLock):
         """Uses the :func:`msvcrt.locking` function to hard lock the lock file on windows systems."""
 
         def _acquire(self) -> None:
-            raise_on_exist_ro_file(self._lock_file)
+            raise_on_not_writable_file(self._lock_file)
             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
+                | os.O_TRUNC  # truncate file if not empty
             )
             try:
                 fd = os.open(self._lock_file, flags, self._mode)
             except OSError as exception:
-                if exception.errno == ENOENT:  # No such file or directory
+                if exception.errno != EACCES:  # has no access to this lock
                     raise
             else:
                 try:
                     msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
-                except OSError:
-                    os.close(fd)
+                except OSError as exception:
+                    os.close(fd)  # close file first
+                    if exception.errno != EACCES:  # file is already locked
+                        raise
                 else:
                     self._lock_file_fd = fd
 

tests/test_filelock.py

@@ -112,12 +112,36 @@ def test_ro_file(lock_type: type[BaseFileLock], tmp_file_ro: Path) -> None:
         lock.acquire()
 
 
+WindowsOnly = pytest.mark.skipif(sys.platform != "win32", reason="Windows only")
+
+
 @pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock])
-def test_missing_directory(lock_type: type[BaseFileLock], tmp_path_ro: Path) -> None:
-    lock_path = tmp_path_ro / "a" / "b"
-    lock = lock_type(str(lock_path))
+@pytest.mark.parametrize(
+    ("expected_error", "match", "bad_lock_file"),
+    [
+        pytest.param(FileNotFoundError, "No such file or directory:", "a/b", id="non_existent_directory"),
+        pytest.param(FileNotFoundError, "No such file or directory:", "", id="blank_filename"),
+        pytest.param(ValueError, "embedded null (byte|character)", "\0", id="null_byte"),
+        pytest.param(
+            PermissionError if sys.platform == "win32" else IsADirectoryError,
+            "Permission denied:" if sys.platform == "win32" else "Is a directory",
+            ".",
+            id="current_directory",
+        ),
+    ]
+    + [pytest.param(OSError, "Invalid argument", i, id=f"invalid_{i}", marks=WindowsOnly) for i in '<>:"|?*\a']
+    + [pytest.param(PermissionError, "Permission denied:", i, id=f"permission_{i}", marks=WindowsOnly) for i in "/\\"],
+)
+@pytest.mark.timeout(5)  # timeout in case of infinite loop
+def test_bad_lock_file(
+    lock_type: type[BaseFileLock],
+    expected_error: type[Exception],
+    match: str,
+    bad_lock_file: str,
+) -> None:
+    lock = lock_type(bad_lock_file)
 
-    with pytest.raises(OSError, match="No such file or directory:"):
+    with pytest.raises(expected_error, match=match):
         lock.acquire()
 
 

whitelist.txt

@@ -6,6 +6,7 @@ caplog
 contnode
 docutils
 eacces
+eisdir
 enosys
 extlinks
 fchmod