Method to list active pg locks

Author: AlanCoding

ansible_base/lib/utils/db.py

@@ -6,9 +6,10 @@
 
 import psycopg
 from django.conf import settings
-from django.db import DEFAULT_DB_ALIAS, OperationalError, connection, connections, transaction
+from django.db import DEFAULT_DB_ALIAS, OperationalError, connection, transaction
 from django.db.backends.postgresql.base import DatabaseWrapper as PsycopgDatabaseWrapper
 from django.db.migrations.executor import MigrationExecutor
+from django.db.transaction import get_connection
 
 logger = logging.getLogger(__name__)
 
@@ -40,6 +41,97 @@ def migrations_are_complete() -> bool:
 # that was licensed under the MIT license
 
 
+def string_to_advisory_lock_id(lock_string: str) -> int:
+    """Convert a string to a PostgreSQL advisory lock ID integer.
+
+    Generates an id within postgres integer range (-2^31 to 2^31 - 1).
+    crc32 generates an unsigned integer in Py3, we convert it into
+    a signed integer using 2's complement (this is a noop in Py2).
+
+    Args:
+        lock_string: The string to convert to a lock ID
+
+    Returns:
+        Integer lock ID suitable for PostgreSQL advisory locks
+    """
+    pos = crc32(lock_string.encode("utf-8"))
+    lock_id = (2**31 - 1) & pos
+    if pos & 2**31:
+        lock_id -= 2**31
+    return lock_id
+
+
+def advisory_lock_id_to_debug_info(lock_id: int) -> dict:
+    """Convert an advisory lock ID back to debug information.
+
+    Note: This cannot reverse the string due to CRC32 hash collisions,
+    but provides debugging information about the lock ID.
+
+    Args:
+        lock_id: The integer lock ID
+
+    Returns:
+        Dictionary with debug information about the lock ID
+    """
+    # Convert back to unsigned for analysis
+    if lock_id < 0:
+        unsigned_value = lock_id + 2**31
+        had_high_bit = True
+    else:
+        unsigned_value = lock_id
+        had_high_bit = False
+
+    return {
+        'lock_id': lock_id,
+        'unsigned_crc32': unsigned_value,
+        'had_high_bit_set': had_high_bit,
+        'hex_representation': hex(lock_id),
+    }
+
+
+def get_active_advisory_locks(using=None) -> list:
+    """Get a list of all currently held advisory locks.
+
+    Args:
+        using: Database alias to use (defaults to DEFAULT_DB_ALIAS)
+
+    Returns:
+        List of dictionaries containing lock information
+    """
+    if using is None:
+        using = DEFAULT_DB_ALIAS
+
+    conn = get_connection(using)
+    if conn.vendor != "postgresql":
+        return []
+
+    with conn.cursor() as cursor:
+        cursor.execute(
+            """
+            SELECT locktype, classid, objid, objsubid, pid, mode, granted
+            FROM pg_locks
+            WHERE locktype = 'advisory'
+        """
+        )
+
+        locks = []
+        for row in cursor.fetchall():
+            locktype, classid, objid, objsubid, pid, mode, granted = row
+            locks.append(
+                {
+                    'locktype': locktype,
+                    'classid': classid,
+                    'objid': objid,
+                    'objsubid': objsubid,
+                    'pid': pid,
+                    'mode': mode,
+                    'granted': granted,
+                }
+            )
+
+        return locks
+
+
 @contextmanager
 def django_pglocks_advisory_lock(lock_id, shared=False, wait=True, using=None):
 
@@ -81,13 +173,7 @@ def django_pglocks_advisory_lock(lock_id, shared=False, wait=True, using=None):
 
         tuple_format = True
     elif isinstance(lock_id, str):
-        # Generates an id within postgres integer range (-2^31 to 2^31 - 1).
-        # crc32 generates an unsigned integer in Py3, we convert it into
-        # a signed integer using 2's complement (this is a noop in Py2)
-        pos = crc32(lock_id.encode("utf-8"))
-        lock_id = (2**31 - 1) & pos
-        if pos & 2**31:
-            lock_id -= 2**31
+        lock_id = string_to_advisory_lock_id(lock_id)
     elif not isinstance(lock_id, int):
         raise ValueError("Cannot use %s as a lock id" % lock_id)
 
@@ -104,7 +190,7 @@ def django_pglocks_advisory_lock(lock_id, shared=False, wait=True, using=None):
     acquire_params = (function_name,) + params
 
     command = base % acquire_params
-    cursor = connections[using].cursor()
+    cursor = get_connection(using).cursor()
 
     cursor.execute(command)
 

test_app/tests/lib/utils/test_db.py

@@ -10,11 +10,14 @@
 
 from ansible_base.lib.utils.db import (
     advisory_lock,
+    advisory_lock_id_to_debug_info,
+    get_active_advisory_locks,
     get_pg_notify_params,
     migrations_are_complete,
     psycopg_conn_string_from_settings_dict,
     psycopg_connection_from_django,
     psycopg_kwargs_from_settings_dict,
+    string_to_advisory_lock_id,
 )
 
 
@@ -148,6 +151,118 @@ def test_psycopg_connection_from_django_new_conn(self):
         assert isinstance(psycopg_connection_from_django(), psycopg.Connection)
 
 
+class TestStringToAdvisoryLockId:
+    """Test the string to advisory lock ID conversion function.
+
+    Generated by Claude Code (Sonnet 4)
+    """
+
+    def test_string_to_advisory_lock_id_basic(self):
+        """Test basic string to lock ID conversion."""
+        lock_id = string_to_advisory_lock_id("test_string")
+        assert isinstance(lock_id, int)
+        assert -(2**31) <= lock_id <= 2**31 - 1
+
+    def test_string_to_advisory_lock_id_consistency(self):
+        """Test that the same string always produces the same lock ID."""
+        test_string = "consistent_test"
+        lock_id1 = string_to_advisory_lock_id(test_string)
+        lock_id2 = string_to_advisory_lock_id(test_string)
+        assert lock_id1 == lock_id2
+
+    def test_string_to_advisory_lock_id_different_strings(self):
+        """Test that different strings produce different lock IDs."""
+        lock_id1 = string_to_advisory_lock_id("string1")
+        lock_id2 = string_to_advisory_lock_id("string2")
+        assert lock_id1 != lock_id2
+
+    def test_string_to_advisory_lock_id_unicode(self):
+        """Test string to lock ID conversion with unicode characters."""
+        lock_id = string_to_advisory_lock_id("test_🔒_unicode")
+        assert isinstance(lock_id, int)
+        assert -(2**31) <= lock_id <= 2**31 - 1
+
+
+class TestAdvisoryLockIdToDebugInfo:
+    """Test the advisory lock ID to debug info function.
+
+    Generated by Claude Code (Sonnet 4)
+    """
+
+    def test_advisory_lock_id_to_debug_info_positive(self):
+        """Test debug info for positive lock ID."""
+        lock_id = 12345
+        debug_info = advisory_lock_id_to_debug_info(lock_id)
+
+        assert debug_info['lock_id'] == lock_id
+        assert debug_info['unsigned_crc32'] == lock_id
+        assert debug_info['had_high_bit_set'] is False
+        assert debug_info['hex_representation'] == hex(lock_id)
+
+    def test_advisory_lock_id_to_debug_info_negative(self):
+        """Test debug info for negative lock ID."""
+        lock_id = -12345
+        debug_info = advisory_lock_id_to_debug_info(lock_id)
+
+        assert debug_info['lock_id'] == lock_id
+        assert debug_info['unsigned_crc32'] == lock_id + 2**31
+        assert debug_info['had_high_bit_set'] is True
+        assert debug_info['hex_representation'] == hex(lock_id)
+
+    def test_advisory_lock_id_to_debug_info_roundtrip(self):
+        """Test debug info for a string-generated lock ID."""
+        test_string = "test_debug_roundtrip"
+        lock_id = string_to_advisory_lock_id(test_string)
+        debug_info = advisory_lock_id_to_debug_info(lock_id)
+
+        assert debug_info['lock_id'] == lock_id
+        assert isinstance(debug_info['unsigned_crc32'], int)
+        assert isinstance(debug_info['had_high_bit_set'], bool)
+        assert debug_info['hex_representation'] == hex(lock_id)
+
+
+class TestGetActiveAdvisoryLocks(SkipIfSqlite):
+    """Test the get active advisory locks function.
+
+    Generated by Claude Code (Sonnet 4)
+    """
+
+    @pytest.mark.django_db
+    def test_get_active_advisory_locks_empty(self):
+        """Test getting active locks when none are held."""
+        locks = get_active_advisory_locks()
+        assert isinstance(locks, list)
+        # We can't guarantee no locks since other tests might be running
+
+    @pytest.mark.django_db
+    def test_get_active_advisory_locks_with_lock(self):
+        """Test getting active locks when we hold one."""
+        test_lock_name = "test_get_active_locks"
+
+        with advisory_lock(test_lock_name):
+            locks = get_active_advisory_locks()
+            assert isinstance(locks, list)
+            # Should have at least our lock
+            assert len(locks) >= 1
+
+            # Check that lock entries have expected structure
+            for lock in locks:
+                assert 'locktype' in lock
+                assert 'classid' in lock
+                assert 'objid' in lock
+                assert 'objsubid' in lock
+                assert 'pid' in lock
+                assert 'mode' in lock
+                assert 'granted' in lock
+                assert lock['locktype'] == 'advisory'
+
+    @pytest.mark.django_db
+    def test_get_active_advisory_locks_sqlite_returns_empty(self):
+        """Test that SQLite returns empty list."""
+        # This test will be skipped by SkipIfSqlite for completeness
+        pass
+
+
 class TestAdvisoryLock(SkipIfSqlite):
     THREAD_WAIT_TIME = 0.1