fix capturing of `BTLEManagementError` on pySwitchbot v0.10.1

Pipfile.lock

@@ -30,10 +30,10 @@
         },
         "pyswitchbot": {
             "hashes": [
-                "sha256:5c2102b1c49f7e9b72d6213b9926a38b32897927569385f6e5830d0b1bb6762e",
-                "sha256:f2a80e123cf494a373dba17ba54995f18a024c774cc28f5b9f67a3bf0e08e7b5"
+                "sha256:392f99abe7c1861c5edbefb5430100941714be127dff03677722496c24ea7d96",
+                "sha256:5a32811423e5ba34e0d1a914ea19442a19c977d2bdd140094eb7413701d85f85"
             ],
-            "version": "==0.10.0"
+            "version": "==0.10.1"
         },
         "sanitized-package": {
             "editable": true,

switchbot_mqtt/__init__.py

@@ -23,6 +23,7 @@ import enum
 import json
 import logging
 import pathlib
+import queue
 import re
 import shlex
 import typing
@@ -49,6 +50,20 @@ def _mac_address_valid(mac_address: str) -> bool:
     return _MAC_ADDRESS_REGEX.match(mac_address.lower()) is not None
 
 
+class _QueueLogHandler(logging.Handler):
+    """
+    logging.handlers.QueueHandler drops exc_info
+    """
+
+    # TypeError: 'type' object is not subscriptable
+    def __init__(self, log_queue: "queue.Queue[logging.LogRecord]") -> None:
+        self.log_queue = log_queue
+        super().__init__()
+
+    def emit(self, record: logging.LogRecord) -> None:
+        self.log_queue.put(record)
+
+
 class _MQTTCallbackUserdata:
     # pylint: disable=too-few-public-methods; @dataclasses.dataclass when python_requires>=3.7
     def __init__(
@@ -281,8 +296,21 @@ class _CurtainMotor(_MQTTControlledActor):
         )
 
     def _update_position(self, mqtt_client: paho.mqtt.client.Client) -> None:
+        log_queue = queue.Queue(maxsize=0)  # type: queue.Queue[logging.LogRecord]
+        logging.getLogger("switchbot").addHandler(_QueueLogHandler(log_queue))
         try:
             self._device.update()
+            # pySwitchbot>=v0.10.1 catches bluepy.btle.BTLEManagementError :(
+            # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.1/switchbot/__init__.py#L141
+            while not log_queue.empty():
+                log_record = log_queue.get()
+                if log_record.exc_info:
+                    exc = log_record.exc_info[1]  # type: typing.Optional[BaseException]
+                    if (
+                        isinstance(exc, bluepy.btle.BTLEManagementError)
+                        and exc.emsg == "Permission Denied"
+                    ):
+                        raise exc
         except bluepy.btle.BTLEManagementError as exc:
             if (
                 exc.emsg == "Permission Denied"

tests/test_switchbot_curtain_motor_device_info.py

@@ -22,27 +22,46 @@ import unittest.mock
 
 import bluepy.btle
 import pytest
-
 import switchbot_mqtt
 
 # pylint: disable=protected-access
 
+_LE_ON_PERMISSION_DENIED_ERROR = bluepy.btle.BTLEManagementError(
+    "Failed to execute management command 'le on'",
+    {
+        "rsp": ["mgmt"],
+        "code": ["mgmterr"],
+        "estat": [20],
+        "emsg": ["Permission Denied"],
+    },
+)
+
+
+def test__update_position_le_on_permission_denied_log():  # pySwitchbot>=v0.10.0
+    actor = switchbot_mqtt._CurtainMotor(
+        mac_address="dummy", retry_count=21, password=None
+    )
+    with unittest.mock.patch(
+        "bluepy.btle.Scanner.scan",
+        side_effect=_LE_ON_PERMISSION_DENIED_ERROR,
+    ), unittest.mock.patch.object(
+        actor, "_report_position"
+    ) as report_position_mock, pytest.raises(
+        PermissionError
+    ) as exc_info:
+        actor._update_position(mqtt_client="client")
+    report_position_mock.assert_not_called()
+    assert "sudo setcap cap_net_admin+ep /" in exc_info.exconly()
+    assert exc_info.value.__cause__ == _LE_ON_PERMISSION_DENIED_ERROR
+
 
-def test__update_position_le_on_permission_denied():
+def test__update_position_le_on_permission_denied_exc():  # pySwitchbot<v0.10.1
     actor = switchbot_mqtt._CurtainMotor(
         mac_address="dummy", retry_count=21, password=None
     )
     with unittest.mock.patch(
         "switchbot.SwitchbotCurtain.update",
-        side_effect=bluepy.btle.BTLEManagementError(
-            "Failed to execute management command 'le on'",
-            {
-                "rsp": ["mgmt"],
-                "code": ["mgmterr"],
-                "estat": [20],
-                "emsg": ["Permission Denied"],
-            },
-        ),
+        side_effect=_LE_ON_PERMISSION_DENIED_ERROR,
     ) as update_mock, unittest.mock.patch.object(
         actor, "_report_position"
     ) as report_position_mock, pytest.raises(
@@ -57,7 +76,7 @@ def test__update_position_le_on_permission_denied():
             exc_info.exconly(),
         ).group(1)
     )
-    assert isinstance(exc_info.value.__cause__, bluepy.btle.BTLEManagementError)
+    assert exc_info.value.__cause__ == _LE_ON_PERMISSION_DENIED_ERROR
 
 
 def test__update_position_other_error():