Hotfix: Restore __init__ method; more robust initialization for singleton locks (#338)

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

Author: ethanbb

src/filelock/_api.py

@@ -85,39 +85,32 @@ class BaseFileLock(ABC, contextlib.ContextDecorator):
     def __new__(  # noqa: PLR0913
         cls,
         lock_file: str | os.PathLike[str],
-        timeout: float = -1,
-        mode: int = 0o644,
-        thread_local: bool = True,  # noqa: FBT001, FBT002
+        timeout: float = -1,  # noqa: ARG003
+        mode: int = 0o644,  # noqa: ARG003
+        thread_local: bool = True,  # noqa: FBT001, FBT002, ARG003
         *,
-        blocking: bool = True,
+        blocking: bool = True,  # noqa: ARG003
         is_singleton: bool = False,
         **kwargs: Any,  # capture remaining kwargs for subclasses  # noqa: ARG003, ANN401
     ) -> Self:
         """Create a new lock object or if specified return the singleton instance for the lock file."""
         if not is_singleton:
-            self = super().__new__(cls)
-            self._initialize(lock_file, timeout, mode, thread_local, blocking=blocking, is_singleton=is_singleton)
-            return self
+            return super().__new__(cls)
 
         instance = cls._instances.get(str(lock_file))
         if not instance:
             self = super().__new__(cls)
-            self._initialize(lock_file, timeout, mode, thread_local, blocking=blocking, is_singleton=is_singleton)
             cls._instances[str(lock_file)] = self
             return self
 
-        if timeout != instance.timeout or mode != instance.mode:
-            msg = "Singleton lock instances cannot be initialized with differing arguments"
-            raise ValueError(msg)
-
         return instance  # type: ignore[return-value] # https://github.com/python/mypy/issues/15322
 
     def __init_subclass__(cls, **kwargs: dict[str, Any]) -> None:
         """Setup unique state for lock subclasses."""
         super().__init_subclass__(**kwargs)
         cls._instances = WeakValueDictionary()
 
-    def _initialize(  # noqa: PLR0913
+    def __init__(  # noqa: PLR0913
         self,
         lock_file: str | os.PathLike[str],
         timeout: float = -1,
@@ -143,6 +136,34 @@ def _initialize(  # noqa: PLR0913
             to pass the same object around.
 
         """
+        if is_singleton and hasattr(self, "_context"):
+            # test whether other parameters match existing instance.
+            if not self.is_singleton:
+                msg = "__init__ should only be called on initialized object if it is a singleton"
+                raise RuntimeError(msg)
+
+            params_to_check = {
+                "thread_local": (thread_local, self.is_thread_local()),
+                "timeout": (timeout, self.timeout),
+                "mode": (mode, self.mode),
+                "blocking": (blocking, self.blocking),
+            }
+
+            non_matching_params = {
+                name: (passed_param, set_param)
+                for name, (passed_param, set_param) in params_to_check.items()
+                if passed_param != set_param
+            }
+            if not non_matching_params:
+                return  # bypass initialization because object is already initialized
+
+            # parameters do not match; raise error
+            msg = "Singleton lock instances cannot be initialized with differing arguments"
+            msg += "\nNon-matching arguments: "
+            for param_name, (passed_param, set_param) in non_matching_params.items():
+                msg += f"\n\t{param_name} (existing lock has {set_param} but {passed_param} was passed)"
+            raise ValueError(msg)
+
         self._is_thread_local = thread_local
         self._is_singleton = is_singleton
 

tests/test_filelock.py

@@ -687,9 +687,10 @@ def __init__(  # noqa: PLR0913 Too many arguments to function call (6 > 5)
             mode: int = 0o644,
             thread_local: bool = True,
             my_param: int = 0,
-            **kwargs: dict[str, Any],
+            **kwargs: dict[str, Any],  # noqa: ARG002
         ) -> None:
-            pass
+            super().__init__(lock_file, timeout, mode, thread_local, blocking=True, is_singleton=True)
+            self.my_param = my_param
 
     lock_path = tmp_path / "a"
     MyFileLock(str(lock_path), my_param=1)
@@ -702,9 +703,10 @@ def __init__(  # noqa: PLR0913 Too many arguments to function call (6 > 5)
             mode: int = 0o644,
             thread_local: bool = True,
             my_param: int = 0,
-            **kwargs: dict[str, Any],
+            **kwargs: dict[str, Any],  # noqa: ARG002
         ) -> None:
-            pass
+            super().__init__(lock_file, timeout, mode, thread_local, blocking=True, is_singleton=True)
+            self.my_param = my_param
 
     MySoftFileLock(str(lock_path), my_param=1)
 
@@ -742,12 +744,19 @@ def test_singleton_locks_are_distinct_per_lock_file(lock_type: type[BaseFileLock
 @pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock])
 def test_singleton_locks_must_be_initialized_with_the_same_args(lock_type: type[BaseFileLock], tmp_path: Path) -> None:
     lock_path = tmp_path / "a"
-    lock = lock_type(str(lock_path), is_singleton=True)  # noqa: F841
-
-    with pytest.raises(ValueError, match="Singleton lock instances cannot be initialized with differing arguments"):
-        lock_type(str(lock_path), timeout=10, is_singleton=True)
-    with pytest.raises(ValueError, match="Singleton lock instances cannot be initialized with differing arguments"):
-        lock_type(str(lock_path), mode=0, is_singleton=True)
+    args: dict[str, Any] = {"timeout": -1, "mode": 0o644, "thread_local": True, "blocking": True}
+    alternate_args: dict[str, Any] = {"timeout": 10, "mode": 0, "thread_local": False, "blocking": False}
+
+    lock = lock_type(str(lock_path), is_singleton=True, **args)
+
+    for arg_name in args:
+        general_msg = "Singleton lock instances cannot be initialized with differing arguments"
+        altered_args = args.copy()
+        altered_args[arg_name] = alternate_args[arg_name]
+        with pytest.raises(ValueError, match=general_msg) as exc_info:
+            lock_type(str(lock_path), is_singleton=True, **altered_args)
+        exc_info.match(arg_name)  # ensure specific non-matching argument is included in exception text
+    del lock, exc_info
 
 
 @pytest.mark.skipif(hasattr(sys, "pypy_version_info"), reason="del() does not trigger GC in PyPy")