bug fix combined-lock-locked issue #10843

xarray/backends/locks.py

@@ -233,7 +233,7 @@ def __exit__(self, *args):
             lock.__exit__(*args)
 
     def locked(self):
-        return any(lock.locked for lock in self.locks)
+        return any(lock.locked() for lock in self.locks)
 
     def __repr__(self):
         return f"CombinedLock({list(self.locks)!r})"

xarray/tests/test_backends_locks.py

@@ -3,6 +3,7 @@
 import threading
 
 from xarray.backends import locks
+from xarray.backends.locks import CombinedLock, SerializableLock
 
 
 def test_threaded_lock() -> None:
@@ -13,3 +14,77 @@ def test_threaded_lock() -> None:
 
     lock3 = locks._get_threaded_lock("bar")
     assert lock1 is not lock3
+
+
+def test_combined_lock_locked_returns_false_when_no_locks_acquired() -> None:
+    """CombinedLock.locked() should return False when no locks are held."""
+    lock1 = threading.Lock()
+    lock2 = threading.Lock()
+    combined = CombinedLock([lock1, lock2])
+
+    assert combined.locked() is False
+    assert lock1.locked() is False
+    assert lock2.locked() is False
+
+
+def test_combined_lock_locked_returns_true_when_one_lock_acquired() -> None:
+    """CombinedLock.locked() should return True when any lock is held."""
+    lock1 = threading.Lock()
+    lock2 = threading.Lock()
+    combined = CombinedLock([lock1, lock2])
+
+    lock1.acquire()
+    try:
+        assert combined.locked() is True
+    finally:
+        lock1.release()
+
+    assert combined.locked() is False
+
+
+def test_combined_lock_locked_returns_true_when_all_locks_acquired() -> None:
+    """CombinedLock.locked() should return True when all locks are held."""
+    lock1 = threading.Lock()
+    lock2 = threading.Lock()
+    combined = CombinedLock([lock1, lock2])
+
+    lock1.acquire()
+    lock2.acquire()
+    try:
+        assert combined.locked() is True
+    finally:
+        lock1.release()
+        lock2.release()
+
+    assert combined.locked() is False
+
+
+def test_combined_lock_locked_with_serializable_locks() -> None:
+    """CombinedLock.locked() should work with SerializableLock instances."""
+    lock1 = SerializableLock()
+    lock2 = SerializableLock()
+    combined = CombinedLock([lock1, lock2])
+
+    assert combined.locked() is False
+
+    lock1.acquire()
+    try:
+        assert combined.locked() is True
+    finally:
+        lock1.release()
+
+    assert combined.locked() is False
+
+
+def test_combined_lock_locked_with_context_manager() -> None:
+    """CombinedLock.locked() should reflect state when using context manager."""
+    lock1 = threading.Lock()
+    lock2 = threading.Lock()
+    combined = CombinedLock([lock1, lock2])
+
+    assert combined.locked() is False
+
+    with combined:
+        assert combined.locked() is True
+
+    assert combined.locked() is False