fix: detect shared-key notifications (call to_string())

at_client/connections/atmonitorconnection.py

@@ -113,7 +113,16 @@ def stop_monitor(self):
 
         self._last_heartbeat_sent_time = self._last_heartbeat_ack_time = TimeUtil.current_time_millis()
         self.disconnect()
-
+
+    @staticmethod
+    def _is_shared_key_notification(atsign, key):
+        """True if `key` is an incoming shared-key notification for `atsign`.
+
+        e.g. "@alice:shared_key@bob" when we (alice) are the receiver. Uses
+        to_string() — calling the bound method, not stringifying it.
+        """
+        return key.startswith(atsign.to_string() + ":shared_key@")
+
     def _run(self):
         what = ""
         first = True
@@ -164,7 +173,7 @@ def _run(self):
                         if uuid == "-1":
                             event_type = AtEventType.STATS_NOTIFICATION
                         elif operation == "update":
-                            if key.startswith(str(self.atsign.to_string) + ":shared_key@"):
+                            if self._is_shared_key_notification(self.atsign, key):
                                 event_type = AtEventType.SHARED_KEY_NOTIFICATION
                             else:
                                 event_type = AtEventType.UPDATE_NOTIFICATION

test/monitor_shared_key_test.py

@@ -0,0 +1,27 @@
+import unittest
+
+from at_client.common import AtSign
+from at_client.connections.atmonitorconnection import AtMonitorConnection
+
+
+class MonitorSharedKeyDetectionTest(unittest.TestCase):
+    """Network-free test for shared-key notification classification.
+
+    Regression for the missing () in `str(self.atsign.to_string)`, which made the
+    check compare against a bound-method repr and never match, so shared-key
+    notifications were mis-typed as ordinary UPDATE notifications.
+    """
+
+    def test_detects_incoming_shared_key_notification(self):
+        me = AtSign("@alice")
+        key = "@alice:shared_key@bob"  # bob sharing his key with me (alice)
+        self.assertTrue(AtMonitorConnection._is_shared_key_notification(me, key))
+
+    def test_ignores_regular_update_notification(self):
+        me = AtSign("@alice")
+        key = "@alice:live_traffic.demo@bob"
+        self.assertFalse(AtMonitorConnection._is_shared_key_notification(me, key))
+
+
+if __name__ == "__main__":
+    unittest.main()