add devices-l command for device paths

adbutils/__init__.py

@@ -33,21 +33,28 @@ def shell(self,
               timeout: typing.Optional[float] = None) -> typing.Union[str, AdbConnection]:
         return self.device(serial).shell(command, stream=stream, timeout=timeout)
 
-    def list(self) -> typing.List[AdbDeviceInfo]:
+    def list(self, extended=False) -> typing.List[AdbDeviceInfo]:
         """
         Returns:
             list of device info, including offline
         """
         infos = []
         with self.make_connection() as c:
-            c.send_command("host:devices")
+            if extended:
+                c.send_command("host:devices-l")
+            else:
+                c.send_command("host:devices")
             c.check_okay()
             output = c.read_string_block()
             for line in output.splitlines():
-                parts = line.strip().split("\t")
-                if len(parts) != 2:
+                parts = line.split()
+                tags = {}
+                num_required_fields = 2 # serial and state
+                if len(parts) < num_required_fields:
                     continue
-                infos.append(AdbDeviceInfo(serial=parts[0], state=parts[1]))
+                if extended:
+                    tags = {**tags, **{kv[0]: kv[1] for kv in list(map(lambda pair: pair.split(":"), parts[num_required_fields:]))}}
+                infos.append(AdbDeviceInfo(serial=parts[0], state=parts[1], tags=tags))
         return infos
 
     def iter_device(self) -> typing.Iterator[AdbDevice]:

adbutils/__main__.py

@@ -95,6 +95,9 @@ def main():
                         "--list",
                         action="store_true",
                         help="list devices")
+    parser.add_argument("--list-extended",
+                        action="store_true",
+                        help="list devices with props (overrides --list)")
     parser.add_argument("-i",
                         "--install",
                         help="install from local apk or url")
@@ -149,6 +152,18 @@ def main():
         print("ADB Server version: {}".format(adbclient.server_version()))
         return
 
+    if args.list_extended:
+        rows = []
+        for info in adbclient.list(extended=True):
+            rows.append([info.serial, " ".join([k+":"+v for (k,v) in info.tags.items()])])
+        lens = []
+        for col in zip(*rows):
+            lens.append(max([len(v) for v in col]))
+        format = "  ".join(["{:<" + str(l) + "}" for l in lens])
+        for row in rows:
+            print(format.format(*row))
+        return
+
     if args.list:
         rows = []
         for d in adbclient.device_list():

adbutils/_proto.py

@@ -13,7 +13,7 @@
 import datetime
 import pathlib
 from typing import List, NamedTuple, Optional, Union
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 
 class Network(str, enum.Enum):
@@ -121,6 +121,7 @@ class ShellReturn:
 class AdbDeviceInfo:
     serial: str
     state: str
+    tags: dict[str] = field(default_factory={})
 
 
 StrOrPathLike = Union[str, pathlib.Path]

tests/adb_server.py

@@ -100,6 +100,35 @@ async def host_list_forward(ctx: Context):
     await ctx.send(encode_string("123456 tcp:1234 tcp:4321"))
 
 
+def enable_devices():
+    @register_command("host:devices")
+    async def host_devices(ctx: Context):
+        await ctx.send(b"OKAY")
+        await ctx.send(encode_string("dummydevice\tdevice\n"))
+
+    @register_command("host:devices-l")
+    async def host_devices_extended(ctx: Context):
+        await ctx.send(b"OKAY")
+        await ctx.send(encode_string("dummydevice\tdevice product:test_emu model:test_model device:test_device\n"))
+
+
+def invalidate_devices():
+    @register_command("host:devices")
+    async def host_devices(ctx: Context):
+        await ctx.send(b"OKAY")
+        await ctx.send(encode_string("dummydevice"))
+
+    @register_command("host:devices-l")
+    async def host_devices_extended(ctx: Context):
+        """"""
+        await ctx.send(b"OKAY")
+        await ctx.send(encode_string("dummydevice"))
+
+SHELL_DEBUGS = {
+    "enable-devices": enable_devices,
+    "invalidate-devices": invalidate_devices
+}
+
 SHELL_OUTPUTS = {
     "pwd": "/",
 }
@@ -123,11 +152,13 @@ async def host_tport_serial(ctx: Context):
     shell_cmd = cmd.split(":", 1)[1]
     if shell_cmd in SHELL_OUTPUTS:
         await ctx.send((SHELL_OUTPUTS[shell_cmd].rstrip() + "\n").encode())
+    elif shell_cmd in SHELL_DEBUGS:
+        SHELL_DEBUGS[shell_cmd]()
+        await ctx.send(b"debug command executed")
     else:
         await ctx.send(b"unknown command")
 
 
-
 async def handle_command(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, server: "AdbServer"):
     try:
          # Receive the command from the client

tests/test_adb_server.py

@@ -27,7 +27,3 @@ def test_host_tport_serial(adb: adbutils.AdbClient):
 
     d = adb.device(serial="123456")
     d.open_transport()
-
-
-
-

tests/test_devices.py

@@ -0,0 +1,34 @@
+# coding: utf-8
+#
+
+
+import pytest
+import adbutils
+from adbutils._proto import AdbDeviceInfo
+
+def test_host_devices(adb: adbutils.AdbClient):
+    _dev = adb.device("any")
+    assert _dev.shell(cmdargs="enable-devices") == 'debug command executed'
+    devices = adb.list(extended=False)
+    assert devices == [AdbDeviceInfo(serial="dummydevice", state="device", tags={})]
+
+
+def test_host_devices_invalid(adb: adbutils.AdbClient):
+    _dev = adb.device("any")
+    assert _dev.shell(cmdargs="invalidate-devices") == 'debug command executed'
+    devices = adb.list(extended=False)
+    assert devices == []
+
+
+def test_host_devices_extended(adb: adbutils.AdbClient):
+    _dev = adb.device("any")
+    assert _dev.shell(cmdargs="enable-devices") == 'debug command executed'
+    devices = adb.list(extended=True)
+    assert devices == [AdbDeviceInfo(serial="dummydevice", state="device", tags={"product": "test_emu", "model": "test_model", "device": "test_device"})]
+
+
+def test_host_devices_extended_invalid(adb: adbutils.AdbClient):
+    _dev = adb.device("any")
+    assert _dev.shell(cmdargs="invalidate-devices") == 'debug command executed'
+    devices = adb.list(extended=True)
+    assert devices == []