notification: Refactor process_receiver_notification

Remove repeated code pattern with generalized implementation. Aim
towards easy extension and code readability.

Related #2273
This commit is contained in:
MattHag 2024-12-30 23:43:24 +01:00 committed by Peter F. Patel-Schneider
parent fa3a9bc5b3
commit 15aaba2802
2 changed files with 58 additions and 12 deletions

View File

@ -48,6 +48,8 @@ if typing.TYPE_CHECKING:
logger = logging.getLogger(__name__)
NotificationHandler = typing.Callable[["Receiver", "HIDPPNotification"], bool]
_hidpp10 = hidpp10.Hidpp10()
_hidpp20 = hidpp20.Hidpp20()
_F = hidpp20_constants.SupportedFeature
@ -68,18 +70,20 @@ def process(device: Device | Receiver, notification: HIDPPNotification):
def process_receiver_notification(receiver: Receiver, hidpp_notification: HIDPPNotification) -> bool | None:
"""Process event messages from receivers."""
if hidpp_notification.sub_id == Notification.PAIRING_LOCK:
return handle_pairing_lock(receiver, hidpp_notification)
elif hidpp_notification.sub_id == Registers.DISCOVERY_STATUS_NOTIFICATION: # Bolt pairing
return handle_discovery_status(receiver, hidpp_notification)
elif hidpp_notification.sub_id == Registers.DEVICE_DISCOVERY_NOTIFICATION: # Bolt pairing
return handle_device_discovery(receiver, hidpp_notification)
elif hidpp_notification.sub_id == Registers.PAIRING_STATUS_NOTIFICATION: # Bolt pairing
return handle_pairing_status(receiver, hidpp_notification)
elif hidpp_notification.sub_id == Registers.PASSKEY_REQUEST_NOTIFICATION: # Bolt pairing
return handle_passkey_request(receiver, hidpp_notification)
elif hidpp_notification.sub_id == Registers.PASSKEY_PRESSED_NOTIFICATION: # Bolt pairing
return handle_passkey_pressed(receiver, hidpp_notification)
event_handler_mapping: dict[int, NotificationHandler] = {
Notification.PAIRING_LOCK: handle_pairing_lock,
Registers.DEVICE_DISCOVERY_NOTIFICATION: handle_device_discovery,
Registers.DISCOVERY_STATUS_NOTIFICATION: handle_discovery_status,
Registers.PAIRING_STATUS_NOTIFICATION: handle_pairing_status,
Registers.PASSKEY_PRESSED_NOTIFICATION: handle_passkey_pressed,
Registers.PASSKEY_REQUEST_NOTIFICATION: handle_passkey_request,
}
try:
handler_func = event_handler_mapping[hidpp_notification.sub_id]
return handler_func(receiver, hidpp_notification)
except KeyError:
pass
assert hidpp_notification.sub_id in [
Notification.CONNECT_DISCONNECT,

View File

@ -277,3 +277,45 @@ def test_process_feature_notification(mocker, hidpp_notification, feature):
result = notifications._process_feature_notification(fake_device, hidpp_notification, feature)
assert result is True
def test_process_receiver_notification_invalid(mocker):
invalid_sub_id = 0x30
notification_data = b"\x02"
notification = HIDPPNotification(0, 0, invalid_sub_id, 0, notification_data)
mock_receiver = mocker.Mock()
with pytest.raises(AssertionError):
notifications.process_receiver_notification(mock_receiver, notification)
def test_handle_device_discovery():
receiver: Receiver = Receiver(MockLowLevelInterface(), None, {}, True, None, None)
sub_id = Registers.DISCOVERY_STATUS_NOTIFICATION
data = b"\x01\x02\x03\x04\x05\x06"
notification = HIDPPNotification(0, 0, sub_id, 0, data)
result = notifications.handle_device_discovery(receiver, notification)
assert result
def test_handle_passkey_request(mocker):
receiver_mock = mocker.Mock()
data = b"\x01"
notification = HIDPPNotification(0, 0, 0, 0, data)
result = notifications.handle_passkey_request(receiver_mock, notification)
assert result is True
def test_handle_passkey_pressed(mocker):
receiver = mocker.Mock()
sub_id = Registers.DISCOVERY_STATUS_NOTIFICATION
data = b"\x01\x02\x03\x04\x05\x06"
notification = HIDPPNotification(0, 0, sub_id, 0, data)
result = notifications.handle_passkey_pressed(receiver, notification)
assert result is True