Queue async ADB commands instead of dropping them during busy operations

.github/workflows/python-package.yml

@@ -15,7 +15,7 @@ env:
 jobs:
   build:
 
-    runs-on: ubuntu-latest
+    runs-on: ${{ matrix.python-version == '3.7' && 'ubuntu-22.04' || 'ubuntu-latest' }}
     strategy:
       matrix:
         python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', '3.12']

androidtv/adb_manager/adb_manager_async.py

@@ -20,7 +20,6 @@
 from ..constants import (
     DEFAULT_ADB_TIMEOUT_S,
     DEFAULT_AUTH_TIMEOUT_S,
-    DEFAULT_LOCK_TIMEOUT_S,
     DEFAULT_TRANSPORT_TIMEOUT_S,
 )
 from ..exceptions import LockNotAcquiredException
@@ -140,15 +139,16 @@ async def device(self, serial):
 
 
 @asynccontextmanager
-async def _acquire(lock, timeout=DEFAULT_LOCK_TIMEOUT_S):
-    """Handle acquisition and release of an ``asyncio.Lock`` object with a timeout.
+async def _acquire(lock, timeout=None):
+    """Queue access to an ``asyncio.Lock``, with an optional acquisition timeout.
 
     Parameters
     ----------
     lock : asyncio.Lock
         The lock that we will try to acquire
-    timeout : float
-        The timeout in seconds
+    timeout : float, None
+        Optional acquisition timeout in seconds. Commands wait in order by default;
+        the ADB transport retains its own timeout.
 
     Yields
     ------

tests/test_adb_manager_async.py

@@ -195,6 +195,41 @@ async def test_adb_shell_lock_not_acquired_not_released(self):
 
                 release.assert_not_called()
 
+    @awaiter
+    async def test_adb_shell_waits_for_busy_connection(self):
+        """A valid command must survive a screenshot holding the lock over 3 s."""
+        with async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell("TEST")[self.PATCH_KEY]:
+            self.assertTrue(await self.adb.connect())
+            lock = asyncio.Lock()
+            await lock.acquire()
+            with patch.object(self.adb, "_adb_lock", lock):
+                command = asyncio.create_task(self.adb.shell("TEST"))
+                try:
+                    await asyncio.sleep(3.05)
+                    self.assertFalse(command.done())
+                finally:
+                    lock.release()
+                    result = await command
+                self.assertEqual(result, "TEST")
+                self.assertFalse(lock.locked())
+
+    @awaiter
+    async def test_cancelled_waiter_does_not_release_busy_connection(self):
+        """Cancelling a queued command must not unlock another command."""
+        with async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell("TEST")[self.PATCH_KEY]:
+            self.assertTrue(await self.adb.connect())
+            lock = asyncio.Lock()
+            await lock.acquire()
+            with patch.object(self.adb, "_adb_lock", lock):
+                command = asyncio.create_task(self.adb.shell("TEST"))
+                await asyncio.sleep(0)
+                command.cancel()
+                with self.assertRaises(asyncio.CancelledError):
+                    await command
+                self.assertTrue(lock.locked())
+                lock.release()
+                self.assertEqual(await self.adb.shell("TEST"), "TEST")
+
     @awaiter
     async def test_adb_push_fail(self):
         """Test when an ADB push command is not executed because the device is unavailable."""