receiver: record passkey entry progress from the receiver

The receiver already sends a PASSKEY_PRESSED notification for every accepted
key or button press during pairing, but the handler discarded all of them.

Count them instead: address 0x00 marks the start of entry, 0x01 one accepted
press, and 0x04 the terminator. Any other address is left alone, since the
rest of that address space is undocumented and a miscounted press is worse
than no count at all.

The notification carries no indication of which button was pressed, so only
the number of accepted presses can be derived from it.

Deliberately does not call receiver.changed(): that path drives the tray, the
main window and desktop notifications, and would fire once per press.

lib/logitech_receiver/notifications.py

@@ -510,6 +510,8 @@ def handle_discovery_status(receiver: Receiver, notification: HIDPPNotification)
             receiver.pairing.counter = receiver.pairing.device_address = None
             receiver.pairing.device_authentication = receiver.pairing.device_name = None
         receiver.pairing.device_passkey = None
+        receiver.pairing.passkey_entered = 0
+        receiver.pairing.passkey_complete = False
         discover_error = ord(notification.data[:1])
         if discover_error:
             receiver.pairing.error = discover_string = hidpp10_constants.BoltPairingError(discover_error).label
@@ -538,6 +540,8 @@ def handle_device_discovery(receiver: Receiver, notification: HIDPPNotification)
 def handle_pairing_status(receiver: Receiver, notification: HIDPPNotification) -> bool:
     with notification_lock:
         receiver.pairing.device_passkey = None
+        receiver.pairing.passkey_entered = 0
+        receiver.pairing.passkey_complete = False
         receiver.pairing.lock_open = notification.address == 0x00
         reason = _("pairing lock is open") if receiver.pairing.lock_open else _("pairing lock is closed")
         if logger.isEnabledFor(logging.INFO):
@@ -564,8 +568,26 @@ def handle_pairing_status(receiver: Receiver, notification: HIDPPNotification) -
 def handle_passkey_request(receiver: Receiver, notification: HIDPPNotification) -> bool:
     with notification_lock:
         receiver.pairing.device_passkey = notification.data[0:6].decode("utf-8")
+        # a fresh passkey means entry is starting over, so any earlier progress is stale
+        receiver.pairing.passkey_entered = 0
+        receiver.pairing.passkey_complete = False
         return True
 
 
-def handle_passkey_pressed(_receiver: Receiver, _hidpp_notification: HIDPPNotification) -> bool:
-    return True
+def handle_passkey_pressed(receiver: Receiver, notification: HIDPPNotification) -> bool:
+    """Tracks how much of the passkey the receiver has accepted so far.
+
+    The receiver reports that a key or button was pressed, but never which one,
+    so only the number of accepted presses can be derived from these events.
+    """
+    with notification_lock:
+        if notification.address == 0x00:  # passkey entry started
+            receiver.pairing.passkey_entered = 0
+            receiver.pairing.passkey_complete = False
+        elif notification.address == 0x01:  # one more digit or bit was accepted
+            receiver.pairing.passkey_entered += 1
+        elif notification.address == 0x04:  # entry terminated, receiver is verifying
+            receiver.pairing.passkey_complete = True
+        else:  # the rest of the address space is undocumented, so do not guess
+            logger.debug("%s: unknown passkey pressed address %02X: %s", receiver, notification.address, notification)
+        return True

lib/logitech_receiver/receiver.py

@@ -79,6 +79,8 @@ class Pairing:
     device_kind: Optional[int] = None
     device_name: Optional[str] = None
     device_passkey: Optional[str] = None
+    passkey_entered: int = 0
+    passkey_complete: bool = False
     new_device: Optional[Device] = None
     error: Optional[any] = None
 

tests/logitech_receiver/test_notifications.py

@@ -325,14 +325,59 @@ def test_handle_passkey_request(mocker):
     result = notifications.handle_passkey_request(receiver_mock, notification)
 
     assert result is True
+    assert receiver_mock.pairing.passkey_entered == 0
+    assert receiver_mock.pairing.passkey_complete is False
 
 
-def test_handle_passkey_pressed(mocker):
-    receiver = mocker.Mock()
-    sub_id = Registers.DISCOVERY_STATUS_NOTIFICATION
+@pytest.mark.parametrize(
+    "address, presses, expected_entered, expected_complete",
+    [
+        (0x00, 1, 0, False),  # entry started
+        (0x01, 3, 3, False),  # one press accepted per notification
+        (0x04, 1, 0, True),  # entry terminated, receiver is verifying
+        (0x07, 3, 0, False),  # undocumented address, both fields left alone
+    ],
+)
+def test_handle_passkey_pressed(address, presses, expected_entered, expected_complete):
+    receiver: Receiver = Receiver(MockLowLevelInterface(), None, {}, True, None, None)
+    sub_id = Registers.PASSKEY_PRESSED_NOTIFICATION
     data = b"\x01\x02\x03\x04\x05\x06"
-    notification = HIDPPNotification(0, 0, sub_id, 0, data)
+    notification = HIDPPNotification(0, 0, sub_id, address, data)
 
-    result = notifications.handle_passkey_pressed(receiver, notification)
+    for _ in range(presses):
+        assert notifications.handle_passkey_pressed(receiver, notification) is True
 
-    assert result is True
+    assert receiver.pairing.passkey_entered == expected_entered
+    assert receiver.pairing.passkey_complete is expected_complete
+
+
+def test_handle_passkey_pressed_does_not_notify(mocker):
+    """Entry progress must not reach the tray, main window or desktop notifications."""
+    receiver: Receiver = Receiver(MockLowLevelInterface(), None, {}, True, None, None)
+    spy_changed = mocker.spy(receiver, "changed")
+    notification = HIDPPNotification(0, 0, Registers.PASSKEY_PRESSED_NOTIFICATION, 0x01, b"\x00" * 6)
+
+    notifications.handle_passkey_pressed(receiver, notification)
+
+    assert spy_changed.call_count == 0
+
+
+@pytest.mark.parametrize(
+    "handler, sub_id, address, data",
+    [
+        (notifications.handle_pairing_status, Registers.PAIRING_STATUS_NOTIFICATION, 0x00, b"\x00" * 8),
+        (notifications.handle_discovery_status, Registers.DISCOVERY_STATUS_NOTIFICATION, 0x00, b"\x00" * 8),
+    ],
+)
+def test_passkey_progress_reset(handler, sub_id, address, data):
+    """A retry or a second device must never inherit a stale press count."""
+    receiver: Receiver = Receiver(MockLowLevelInterface(), None, {}, True, None, None)
+    receiver.pairing.passkey_entered = 7
+    receiver.pairing.passkey_complete = True
+    notification = HIDPPNotification(0, 0, sub_id, address, data)
+
+    assert handler(receiver, notification) is True
+
+    assert receiver.pairing.device_passkey is None
+    assert receiver.pairing.passkey_entered == 0
+    assert receiver.pairing.passkey_complete is False