enhance file lock management in AppendOnlyFSBucket tests

python/tests/bucket_tester.py

@@ -10,6 +10,7 @@
 from queue import Queue
 from typing import BinaryIO
 from unittest import TestCase
+import uuid
 
 import pyarrow as pa
 import pyarrow.parquet as pq
@@ -87,7 +88,7 @@ def __init__(self, storage: IBucket, test_case: TestCase) -> None:
         self.storage = storage
         self.test_case = test_case
         # Next is a unique suffix to be used in the names of dirs and files, so they will be unique
-        self.us = f"{iTSms.now() % 100_000_000:08d}"
+        self.us = uuid.uuid4().hex
 
     def cleanup(self) -> None:
         self.storage.remove_prefix(f"dir{self.us}")

python/tests/test_append_only_fs_bucket.py

@@ -7,6 +7,7 @@
 
 from bucketbase import MemoryBucket
 from bucketbase.fs_bucket import AppendOnlyFSBucket, FSBucket
+from bucketbase.named_lock_manager import FileLockManager
 
 
 class TestAppendOnlyFSBucket(unittest.TestCase):
@@ -44,8 +45,14 @@ def put_object(self2, name, content):
         # put object is expected to create a lock file before calling base_bucket.put_object, and remove it after
         bucket_in_test.put_object(object_name, content)
 
-        self.assertFalse(lock_file_path.exists())
-        self.assertEqual(base_bucket_put_calls, [(object_name, content)])
+        lock_verifier = FileLockManager(self.locks_path)
+        v_lock = lock_verifier.get_lock(object_name)
+        try:
+            acquired = v_lock.acquire(timeout=0.1)
+            self.assertTrue(acquired, "Lock should have been released after put_object")
+            self.assertEqual(base_bucket_put_calls, [(object_name, content)])
+        finally:
+            v_lock.release()
 
     def test_put_object_twice_raises_exception(self):
         bucket_in_test = AppendOnlyFSBucket(self.base_bucket, self.locks_path)
@@ -86,12 +93,15 @@ def test_lock_object_creates_lock_and_unlock_releases(self):
         # Attempt to lock the object
         bucket_in_test._lock_object(object_name)
 
-        # Check if the lock file was created
-        lock_file_path = self.locks_path / (object_name.replace(bucket_in_test.SEP, FSBucket.TEMP_SEP) + ".lock")
-        self.assertTrue(lock_file_path.exists())
+        # Verify the lock is actually held by attempting acquisition with another manager
+        lock_verifier = FileLockManager(self.locks_path)
+        with self.assertRaises(TimeoutError):
+            lock_verifier.get_lock(object_name).acquire(timeout=0.1)
 
         bucket_in_test._unlock_object(object_name)
-        self.assertFalse(lock_file_path.exists())
+        v_lock = lock_verifier.get_lock(object_name)
+        self.assertTrue(v_lock.acquire(timeout=0.1), "Lock should have been released after _unlock_object")
+        v_lock.release()
 
     def test_unlocking_unlocked_object_raises_assertion(self):
         bucket_in_test = AppendOnlyFSBucket(self.base_bucket, self.locks_path)
@@ -213,3 +223,7 @@ def thread2_func():
         t2.join()
         self.assertListEqual(results, ["thread1_done", "thread2_file_exists"])
         self.assertEqual(bucket_in_test.get_object(object_name), content1)
+
+
+if __name__ == "__main__":
+    unittest.main()

python/tests/test_namedlock.py

@@ -71,16 +71,24 @@ def test_sanitizes_path_characters(self):
         lock1.acquire()
         lock2.acquire()
 
-        # Verify both lock files were created with sanitized names
-        files = list(self.temp_dir.glob("*"))
-        sanitized_names = [f.name for f in files]
-
-        self.assertIn("path#with#slashes.lock", sanitized_names)
-        self.assertIn("path#with#slashes2.lock", sanitized_names)
-        lock1.release()
-        lock2.release()
-
-        self.assertEqual(0, len(list(self.temp_dir.glob("*"))))
+        lock_verifier = FileLockManager(self.temp_dir)
+        try:
+            # Attempting to acquire already-held locks should timeout, raising TimeoutError
+            # (filelock.Timeout is a subclass of TimeoutError).
+            with self.assertRaises(TimeoutError):
+                lock_verifier.get_lock(name1).acquire(timeout=0.1)
+            with self.assertRaises(TimeoutError):
+                lock_verifier.get_lock(name2).acquire(timeout=0.1)
+        finally:
+            lock1.release()
+            lock2.release()
+
+        v_lock1 = lock_verifier.get_lock(name1)
+        self.assertTrue(v_lock1.acquire(timeout=0.1))
+        v_lock1.release()
+        v_lock2 = lock_verifier.get_lock(name2)
+        self.assertTrue(v_lock2.acquire(timeout=0.1))
+        v_lock2.release()
 
     def test_lock_directory_created(self):
         # Delete the directory to test creation
@@ -93,3 +101,18 @@ def test_lock_directory_created(self):
 
         self.assertTrue(self.temp_dir.exists())
         self.assertTrue(self.temp_dir.is_dir())
+
+    def test_released_lock_can_be_reacquired_by_new_manager(self):
+        name = "some_lock"
+        lock1 = self.lock_manager.get_lock(name)
+        lock1.acquire()
+        lock1.release()
+
+        new_manager = FileLockManager(self.temp_dir)
+        lock2 = new_manager.get_lock(name)
+        self.assertTrue(lock2.acquire(timeout=0.1))
+        lock2.release()
+
+
+if __name__ == "__main__":
+    unittest.main()