fix(ndb): avoid unbound lock in delete callback (#17969)

## Summary

This PR fixes a `NameError` that occurs when `ndb.delete()` is called
with `use_datastore=False` inside a transaction while the global cache
is enabled.

The transaction completion callback can reference `lock` even though no
lock is acquired when datastore writes are disabled.

## Changes

- Avoid registering the unlock callback when no lock has been acquired.
- Add a regression test covering this scenario.

## Testing

```bash
python -m pytest tests/unit/test__datastore_api.py -v
```

All 78 tests passed.

Fixes #17926

---------

Co-authored-by: Anthonios Partheniou <partheniou@google.com>

Author: lyushher

packages/google-cloud-ndb/google/cloud/ndb/_datastore_api.py

@@ -434,6 +434,7 @@ def delete(key, options):
     use_global_cache = context._use_global_cache(key, options)
     use_datastore = context._use_datastore(key, options)
     transaction = context.transaction
+    lock = None
 
     if use_global_cache:
         cache_key = _cache.global_cache_key(key)
@@ -450,17 +451,18 @@ def delete(key, options):
         yield batch.delete(key)
 
     if use_global_cache:
-        if transaction:
+        if lock is not None:
+            if transaction:
 
-            def callback():
-                _cache.global_unlock_for_write(cache_key, lock).result()
+                def callback():
+                    _cache.global_unlock_for_write(cache_key, lock).result()
 
-            context.call_on_transaction_complete(callback)
+                context.call_on_transaction_complete(callback)
 
-        elif use_datastore:
-            yield _cache.global_unlock_for_write(cache_key, lock)
+            else:
+                yield _cache.global_unlock_for_write(cache_key, lock)
 
-        else:
+        elif not use_datastore:
             yield _cache.global_delete(cache_key)
 
 

packages/google-cloud-ndb/tests/unit/test__datastore_api.py

@@ -882,6 +882,26 @@ def test_cache_enabled(Batch, global_cache):
 
         assert not global_cache.get([cache_key])[0]
 
+    @staticmethod
+    @mock.patch("google.cloud.ndb._datastore_api._NonTransactionalCommitBatch")
+    def test_lock_not_acquired(Batch, global_cache):
+        key = key_module.Key("SomeKind", 1)
+        cache_key = _cache.global_cache_key(key._key)
+        global_cache.set({cache_key: b"foo"})
+
+        batch = Batch.return_value
+        batch.delete.return_value = future_result(None)
+
+        with mock.patch(
+            "google.cloud.ndb._datastore_api._cache.global_lock_for_write",
+            return_value=future_result(None),
+        ):
+            future = _api.delete(key._key, _options.Options())
+
+        assert future.result() is None
+        batch.delete.assert_called_once_with(key._key)
+        assert global_cache.get([cache_key]) == [b"foo"]
+
     @staticmethod
     @mock.patch("google.cloud.ndb._datastore_api._NonTransactionalCommitBatch")
     def test_w_transaction(Batch, global_cache):
@@ -921,6 +941,31 @@ def test_without_datastore(Batch, global_cache):
 
         assert global_cache.get([cache_key]) == [None]
 
+    @staticmethod
+    @mock.patch("google.cloud.ndb._datastore_api._NonTransactionalCommitBatch")
+    def test_without_datastore_with_transaction(Batch, global_cache):
+        context = context_module.get_context()
+        callbacks = []
+
+        with context.new(
+            transaction=b"abc123", transaction_complete_callbacks=callbacks
+        ).use():
+            key = key_module.Key("SomeKind", 1)
+            cache_key = _cache.global_cache_key(key._key)
+            global_cache.set({cache_key: b"foo"})
+
+            batch = Batch.return_value
+            batch.delete.side_effect = Exception("Shouldn't use Datastore")
+
+            future = _api.delete(
+                key._key,
+                _options.Options(use_datastore=False),
+            )
+            assert future.result() is None
+
+            assert callbacks == []
+            assert global_cache.get([cache_key]) == [None]
+
     @staticmethod
     @mock.patch("google.cloud.ndb._datastore_api._NonTransactionalCommitBatch")
     def test_cache_disabled(Batch, global_cache):