Merge pull request #84 from google/gbg/78

use libusb auto-detach feature

.github/workflows/python-build-test.yml

@@ -21,7 +21,7 @@ jobs:
     - name: Get history and tags for SCM versioning to work
       run: |
         git fetch --prune --unshallow
-        git fetch --depth=1 origin +refs/tags/*:refs/tags/*      
+        git fetch --depth=1 origin +refs/tags/*:refs/tags/*
     - name: Set up Python 3.10
       uses: actions/setup-python@v3
       with:
@@ -30,9 +30,9 @@ jobs:
       run: |
         python -m pip install --upgrade pip
         python -m pip install ".[build,test,development,documentation]"
-    - name: Test with pytest
+    - name: Test
       run: |
-        pytest
+        invoke test
     - name: Build
       run: |
         inv build

apps/console.py

@@ -443,7 +443,10 @@ class ConsoleApp:
         # Discover all services, characteristics and descriptors
         self.append_to_output('discovering services...')
         await self.connected_peer.discover_services()
-        self.append_to_output(f'found {len(self.connected_peer.services)} services, discovering charateristics...')
+        self.append_to_output(
+            f'found {len(self.connected_peer.services)} services,'
+            ' discovering characteristics...'
+        )
         await self.connected_peer.discover_characteristics()
         self.append_to_output('found characteristics, discovering descriptors...')
         for service in self.connected_peer.services:

bumble/host.py

@@ -121,24 +121,28 @@ class Host(EventEmitter):
             self.hc_acl_data_packet_length     = response.return_parameters.hc_acl_data_packet_length
             self.hc_total_num_acl_data_packets = response.return_parameters.hc_total_num_acl_data_packets
 
+            logger.debug(
+                f'HCI ACL flow control: hc_acl_data_packet_length={self.hc_acl_data_packet_length},'
+                f'hc_total_num_acl_data_packets={self.hc_total_num_acl_data_packets}'
+            )
+
         if self.supports_command(HCI_LE_READ_BUFFER_SIZE_COMMAND):
             response = await self.send_command(HCI_LE_Read_Buffer_Size_Command(), check_result=True)
             self.hc_le_acl_data_packet_length     = response.return_parameters.hc_le_acl_data_packet_length
             self.hc_total_num_le_acl_data_packets = response.return_parameters.hc_total_num_le_acl_data_packets
 
-        if response.return_parameters.hc_le_acl_data_packet_length == 0 or response.return_parameters.hc_total_num_le_acl_data_packets == 0:
-            # LE and Classic share the same values
-            self.hc_le_acl_data_packet_length     = self.hc_acl_data_packet_length
-            self.hc_total_num_le_acl_data_packets = self.hc_total_num_acl_data_packets
+            logger.debug(
+                f'HCI LE ACL flow control: hc_le_acl_data_packet_length={self.hc_le_acl_data_packet_length},'
+                f'hc_total_num_le_acl_data_packets={self.hc_total_num_le_acl_data_packets}'
+            )
 
-        logger.debug(
-            f'HCI ACL flow control: hc_acl_data_packet_length={self.hc_acl_data_packet_length},'
-            f'hc_total_num_acl_data_packets={self.hc_total_num_acl_data_packets}'
-        )
-        logger.debug(
-            f'HCI LE ACL flow control: hc_le_acl_data_packet_length={self.hc_le_acl_data_packet_length},'
-            f'hc_total_num_le_acl_data_packets={self.hc_total_num_le_acl_data_packets}'
-        )
+            if (
+                response.return_parameters.hc_le_acl_data_packet_length == 0 or
+                response.return_parameters.hc_total_num_le_acl_data_packets == 0
+            ):
+                # LE and Classic share the same values
+                self.hc_le_acl_data_packet_length     = self.hc_acl_data_packet_length
+                self.hc_total_num_le_acl_data_packets = self.hc_total_num_acl_data_packets
 
         if (
             self.supports_command(HCI_LE_READ_SUGGESTED_DEFAULT_DATA_LENGTH_COMMAND) and

bumble/l2cap.py

@@ -1224,7 +1224,7 @@ class ChannelManager:
             self._host.remove_listener('disconnection', self.on_disconnection)
         self._host = host
         if host is not None:
-            host.add_listener('disconnection', self.on_disconnection)
+            host.on('disconnection', self.on_disconnection)
 
     def find_channel(self, connection_handle, cid):
         if connection_channels := self.channels.get(connection_handle):

bumble/rfcomm.py

@@ -636,8 +636,8 @@ class Multiplexer(EventEmitter):
             if self.open_result:
                 self.open_result.set_exception(ConnectionError(
                     ConnectionError.CONNECTION_REFUSED,
-                    self.l2cap_channel.connection.peer_address,
                     BT_BR_EDR_TRANSPORT,
+                    self.l2cap_channel.connection.peer_address,
                     'rfcomm'
                 ))
         else:

bumble/smp.py

@@ -1110,7 +1110,7 @@ class Session:
         self.manager.on_pairing(self, peer_address, keys)
 
     def on_pairing_failure(self, reason):
-        logger.warn(f'pairing failure ({error_name(reason)})')
+        logger.warning(f'pairing failure ({error_name(reason)})')
 
         if self.completed:
             return

bumble/transport/usb.py

@@ -414,14 +414,13 @@ async def open_usb_transport(spec):
 
         device = found.open()
 
-        # Detach the kernel driver if supported and needed
+        # Auto-detach the kernel driver if supported
         if usb1.hasCapability(usb1.CAP_SUPPORTS_DETACH_KERNEL_DRIVER):
             try:
-                if device.kernelDriverActive(interface):
-                    logger.debug("detaching kernel driver")
-                    device.detachKernelDriver(interface)
-            except usb1.USBError:
-                pass
+                logger.debug('auto-detaching kernel driver')
+                device.setAutoDetachKernelDriver(True)
+            except usb1.USBError as error:
+                logger.warning(f'unable to auto-detach kernel driver: {error}')
 
         # Set the configuration if needed
         try:

setup.cfg

@@ -65,6 +65,7 @@ build =
 test =
     pytest >= 6.2
     pytest-asyncio >= 0.17
+    pytest-html >= 3.2.0
     coverage >= 6.4
 development =
     invoke >= 1.4

tasks.py

@@ -53,7 +53,7 @@ test_tasks = Collection()
 ns.add_collection(test_tasks, name="test")
 
 @task
-def test(ctx, filter=None, junit=False, install=False):
+def test(ctx, filter=None, junit=False, install=False, html=False):
     # Install the package before running the tests
     if install:
         ctx.run("python -m pip install .[test]")
@@ -63,6 +63,8 @@ def test(ctx, filter=None, junit=False, install=False):
         args += "--junit-xml test-results.xml"
     if filter is not None:
         args += " -k '{}'".format(filter)
+    if html:
+        args += "--html results.html"
     ctx.run("python -m pytest {} {}".format(os.path.join(ROOT_DIR, "tests"), args))
 
 test_tasks.add_task(test, default=True)

tests/import_test.py

@@ -64,37 +64,37 @@ def test_import():
 
 # -----------------------------------------------------------------------------
 def test_app_imports():
-    from bumble.apps.console import main
+    from apps.console import main
     assert main
 
-    from bumble.apps.controller_info import main
+    from apps.controller_info import main
     assert main
 
-    from bumble.apps.controllers import main
+    from apps.controllers import main
     assert main
 
-    from bumble.apps.gatt_dump import main
+    from apps.gatt_dump import main
     assert main
 
-    from bumble.apps.gg_bridge import main
+    from apps.gg_bridge import main
     assert main
 
-    from bumble.apps.hci_bridge import main
+    from apps.hci_bridge import main
     assert main
 
-    from bumble.apps.pair import main
+    from apps.pair import main
     assert main
 
-    from bumble.apps.scan import main
+    from apps.scan import main
     assert main
 
-    from bumble.apps.show import main
+    from apps.show import main
     assert main
 
-    from bumble.apps.unbond import main
+    from apps.unbond import main
     assert main
 
-    from bumble.apps.usb_probe import main
+    from apps.usb_probe import main
     assert main