Method acquire_and_capture

tests/unit/test_session_lock.py:362–371

()

            try:

                async def acquire_and_capture():
                    # Get the per-loop manager
                    per_loop_mgr = manager._get_loop_manager()
                    # Capture the lock object id before acquiring
                    async with per_loop_mgr._access_lock:
                        lock = per_loop_mgr._locks[session_id]
                        with lock_id_lock:
                            lock_ids.add(id(lock))
                    async with manager.acquire_lock(session_id):
                        await asyncio.sleep(0.01)

                loop.run_until_complete(acquire_and_capture())
            finally: