spack.util.lock: add type-hints, remove **kwargs in method signatures (#39011)

lib/spack/llnl/util/lock.py

@@ -214,6 +214,7 @@ class Lock:
     def __init__(
         self,
         path: str,
+        *,
         start: int = 0,
         length: int = 0,
         default_timeout: Optional[float] = None,

lib/spack/spack/stage.py

@@ -337,7 +337,7 @@ def __init__(
 
                 tty.debug("Creating stage lock {0}".format(self.name))
                 Stage.stage_locks[self.name] = spack.util.lock.Lock(
-                    stage_lock_path, lock_id, 1, desc=self.name
+                    stage_lock_path, start=lock_id, length=1, desc=self.name
                 )
 
             self._lock = Stage.stage_locks[self.name]

lib/spack/spack/test/llnl/util/lock.py

@@ -307,7 +307,7 @@ def __name__(self):
         return self.__class__.__name__
 
     def __call__(self, barrier):
-        lock = lk.Lock(self.lock_path, self.start, self.length)
+        lock = lk.Lock(self.lock_path, start=self.start, length=self.length)
         lock.acquire_write()  # grab exclusive lock
         barrier.wait()
         barrier.wait()  # hold the lock until timeout in other procs.
@@ -324,7 +324,7 @@ def __name__(self):
         return self.__class__.__name__
 
     def __call__(self, barrier):
-        lock = lk.Lock(self.lock_path, self.start, self.length)
+        lock = lk.Lock(self.lock_path, start=self.start, length=self.length)
         lock.acquire_read()  # grab shared lock
         barrier.wait()
         barrier.wait()  # hold the lock until timeout in other procs.
@@ -341,7 +341,7 @@ def __name__(self):
         return self.__class__.__name__
 
     def __call__(self, barrier):
-        lock = lk.Lock(self.lock_path, self.start, self.length)
+        lock = lk.Lock(self.lock_path, start=self.start, length=self.length)
         barrier.wait()  # wait for lock acquire in first process
         with pytest.raises(lk.LockTimeoutError):
             lock.acquire_write(lock_fail_timeout)
@@ -359,7 +359,7 @@ def __name__(self):
         return self.__class__.__name__
 
     def __call__(self, barrier):
-        lock = lk.Lock(self.lock_path, self.start, self.length)
+        lock = lk.Lock(self.lock_path, start=self.start, length=self.length)
         barrier.wait()  # wait for lock acquire in first process
         with pytest.raises(lk.LockTimeoutError):
             lock.acquire_read(lock_fail_timeout)

lib/spack/spack/util/lock.py

@@ -7,6 +7,7 @@
 import os
 import stat
 import sys
+from typing import Optional, Tuple
 
 import llnl.util.lock
 
@@ -29,36 +30,48 @@ class Lock(llnl.util.lock.Lock):
     the actual locking mechanism can be disabled via ``_enable_locks``.
     """
 
-    def __init__(self, *args, **kwargs):
-        enable_lock = kwargs.pop("enable", None)
+    def __init__(
+        self,
+        path: str,
+        *,
+        start: int = 0,
+        length: int = 0,
+        default_timeout: Optional[float] = None,
+        debug: bool = False,
+        desc: str = "",
+        enable: Optional[bool] = None,
+    ) -> None:
+        enable_lock = enable
         if sys.platform == "win32":
             enable_lock = False
         elif sys.platform != "win32" and enable_lock is None:
             enable_lock = True
         self._enable = enable_lock
-        super(Lock, self).__init__(*args, **kwargs)
+        super().__init__(
+            path,
+            start=start,
+            length=length,
+            default_timeout=default_timeout,
+            debug=debug,
+            desc=desc,
+        )
 
-    def _lock(self, op, timeout=0):
+    def _lock(self, op: int, timeout: Optional[float] = 0.0) -> Tuple[float, int]:
         if self._enable:
             return super()._lock(op, timeout)
-        else:
-            return 0, 0
+        return 0.0, 0
 
-    def _unlock(self):
+    def _unlock(self) -> None:
         """Unlock call that always succeeds."""
         if self._enable:
             super()._unlock()
 
-    def _debug(self, *args):
-        if self._enable:
-            super()._debug(*args)
-
-    def cleanup(self, *args):
+    def cleanup(self, *args) -> None:
         if self._enable:
             super().cleanup(*args)
 
 
-def check_lock_safety(path):
+def check_lock_safety(path: str) -> None:
     """Do some extra checks to ensure disabling locks is safe.
 
     This will raise an error if ``path`` can is group- or world-writable
@@ -82,9 +95,9 @@ def check_lock_safety(path):
             writable = "world"
 
         if writable:
-            msg = "Refusing to disable locks: spack is {0}-writable.".format(writable)
+            msg = f"Refusing to disable locks: spack is {writable}-writable."
             long_msg = (
-                "Running a shared spack without locks is unsafe. You must "
-                "restrict permissions on {0} or enable locks."
-            ).format(path)
+                f"Running a shared spack without locks is unsafe. You must "
+                f"restrict permissions on {path} or enable locks."
+            )
             raise spack.error.SpackError(msg, long_msg)